Correct way to use Dask on a cluster

[GertKl]

Hi,

I have no prior experience using Dask or similar libraries, and am struggling to use the dask simulator on a HPC cluster. To my understanding, it should be as simple as (referring to example notebook 6) exchanging cluster=LocalCluster() with for example cluster=PBSCluster(). However, I only have access to a Torque cluster, to which it seems that dask doesn't have an out-of-the-box scheduler, for example TorqueCluster(). Is there an easy (or less easy) way to still used dask on such a cluster?

I thought that I could at least just use LocalCluster if running the code on a single node, thus exploiting that the node at least has more cores than my laptop. However, when I do this, it seems as though the node never really starts mulitprocessing, which is evident from observing the output from $top, indicating that only ever a total of 100% of CPU power is being used, i.e. only a single process (this seems to be the same issue as brought up by @erinaldi in his discussion thread from may 1st, to which he unfortunately doesn't seem to have received an answer). It is also evident because there is no difference in simulation times, no matter how many workers I choose.

When running any of my codes, including example notebook 6, on my laptop, it is all working as expected.

I would appreciate any help with this, as I am in great need of faster simulation times.

[cweniger]

Hi GertKI, I would recommend to use lightning branch of Swyft, which is the current development branch. It doesn't use dask
anymore for parallel simulations, and is better documented (https://swyft.readthedocs.io/en/lightning/).

As explained here, you can use a ZarrStore and start the simulations in multiple threads, which then will populate the same store in parallel.

[fnattino]

Hi @GertKl! I'll try to answer your question anyway in case you still want to try out the Dask-based implementation from swyft 0.3.2, maybe to make use of multiple nodes of a cluster.

However, I only have access to a Torque cluster, to which it seems that dask doesn't have an out-of-the-box scheduler, for example TorqueCluster(). Is there an easy (or less easy) way to still used dask on such a cluster?

My guess is that the PBSCluster() from Dask Jobqueue should work for TORQUE. After installing dask-jobqueue with either pip or conda (from the conda-forge channel), you should be able to create a Dask cluster using:

from dask_jobqueue import PBSCluster

cluster = PBSCluster(
    queue="thin",
    memory="60GiB",  # max memory per job
    processes=4,  # num processes per job
    cores=4,  # num threads per job 
# resource_spec="nodes=1:ppn=4",  # additionally configure request of resources  
    local_directory="$TMPDIR",
    walltime="01:00:00",
)

# running the following print the job script which is submitted 
# to the PBS scheduler to request one additional worker 
print(cluster.job_script())

# add 5 workers to the Dask cluster => submit 5 jobs to the PBS scheduler
cluster.scale(jobs=5) 

Note that cores must be equal to processes for command-line simulators. This is just for the Dask settings, but you can still run multi-threaded command-line simulators if more cores are requested via a proper setting of resource_spec.

If the PBSCluster does not turn out to work well with your TORQUE instance, an alternative would be to start the cluster "manually", which should be also rather straightforward.

It is also evident because there is no difference in simulation times, no matter how many workers I choose.

Concerning your second issues, have you tried setting batch_size when calling store.simulate() (cell 17 in Notebook 6)? This setting regulates the size of the batches used to submit simulations. The default is that all simulations are run in a single batch (which is thus equivalent to running the simulations sequentially), see also #97. If you have 1000 simulations and 4 cores you might want to try batch_size=250 or smaller (too small batches will lead to large overhead timings).

[GertKl]

Thank you both for your quick replies!

@cweniger, can I expect any improvements in performance from Swyft-lightning? I saw that it was supposed to entail a “complete interface overhaul”, so I have to balance the effort it takes to adapt my code against the potential benefits. At least, for now, assuming that I can easily make an operational analysis using swyft-0.3.2.

I’ve been taking a look at your suggestions, @fnattino. Indeed, using a suitable batch_size parameter value does make the simulation-process behave more like expected. Using the PBS scheduler seems to produce the same results as a LocalCluster, but it prints an error which I'm not sure if I should be worried about:

2022-11-11 14:18:20,439 - tornado.application - ERROR - Exception in callback functools.partial(<bound method IOLoop._discard_future_result of <tornado.platform.asyncio.AsyncIOLoop object at 0x7f0412834b20>>, <Task finished name='Task-21' coro=<SpecCluster._correct_state_internal() done, defined at /home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/distributed/deploy/spec.py:319> exception=RuntimeError('Command exited with non-zero exit code.\nExit code: 172\nCommand:\nqsub /tmp/tmp679j4e_4.sh\nstdout:\n\nstderr:\nqsub: submit error (Unknown queue MSG=requested queue not found)\n\n')>)
Traceback (most recent call last):
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/tornado/ioloop.py", line 741, in _run_callback
    ret = callback()
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/tornado/ioloop.py", line 765, in _discard_future_result
    future.result()
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/distributed/deploy/spec.py", line 358, in _correct_state_internal
    await w  # for tornado gen.coroutine support
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/distributed/deploy/spec.py", line 63, in _
    await self.start()
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/dask_jobqueue/core.py", line 411, in start
    out = await self._submit_job(fn)
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/dask_jobqueue/core.py", line 394, in _submit_job
    return self._call(shlex.split(self.submit_command) + [script_filename])
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/dask_jobqueue/core.py", line 489, in _call
    raise RuntimeError(
RuntimeError: Command exited with non-zero exit code.
Exit code: 172
Command:
qsub /tmp/tmp679j4e_4.sh
stdout:

stderr:
qsub: submit error (Unknown queue MSG=requested queue not found)

This error is repeated the same number of times as the number of processes + 2, it seems.

The last line suggests to me that the scheduler is trying to employ qsub, but cannot, as I am using Torque and not PBS. How much of a problem is this? Should I expect that using it will schedule jobs on the cluster as though I had run them using $qsub? I.e. when calling $qstat, should I see the jobs that are being run by dask (currently this is not the case)?

By the way, I am assuming that my script, which employs dask, should not be run by calling $qsub myself? At least, it only works if I don’t.

[fnattino]

Hey @GertKl, happy it helped!

The snippet I provided above should create a Dask cluster by starting a Dask scheduler process wherever you are executing the script (on the login node? on a compute node of the cluster?) and by submitting jobs to the Torque/PBS scheduler asking for resources to start the workers. These worker jobs are actually submitted when calling cluster.scale(...) and they need to be running for swyft to be able to run simulations via Dask. When worker jobs are running you should see them in the cluster queue by running qstat.

The error you got suggests indeed that Dask is not able to get resources from the Torque scheduler because of a failure in the call to qsub. The error maybe suggests that the name of the queue is not set correctly? You should be able to set the queue where you want to submit worker jobs via the queue input parameter for PBSCluster.

In order to debug the issue further it can be helpful to have a look at the job script that you get when running cluster.job_script():

from dask_jobqueue import PBSCluster
cluster = PBSCluster(...)  # configure the job features depending on your settings/needs
print(cluster.job_script())  

The script printed to stdout is the script that Dask submits in order to request workers - so you should be able to verify whether there are Torque fields that are incompatible with the settings of your cluster.

[GertKl]

Hi again @fnattino, thanks again for your quick help!

Indeed, I had not specified the name of the queue correctly. The simulations are now running as expected, as far as I can see for up to 10000 simulations, but I am having trouble scaling this. Already for 50000 simulations I am getting the following errors:

2022-11-17 14:04:07,854 - distributed.utils_perf - WARNING - full garbage collections took 23% CPU time recently (threshold: 10%)
2022-11-17 14:04:39,381 - distributed.utils_perf - WARNING - full garbage collections took 24% CPU time recently (threshold: 10%)
2022-11-17 14:07:54,932 - distributed.utils_perf - WARNING - full garbage collections took 17% CPU time recently (threshold: 10%)
2022-11-17 14:08:40,249 - distributed.utils_perf - WARNING - full garbage collections took 18% CPU time recently (threshold: 10%)
2022-11-17 14:09:45,088 - distributed.utils_perf - WARNING - full garbage collections took 19% CPU time recently (threshold: 10%)
2022-11-17 14:10:21,130 - distributed.utils_perf - WARNING - full garbage collections took 19% CPU time recently (threshold: 10%)
2022-11-17 14:11:08,822 - distributed.utils_perf - WARNING - full garbage collections took 20% CPU time recently (threshold: 10%)
2022-11-17 14:12:04,561 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:13:12,131 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:18:46,656 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:18:53,831 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:19:11,868 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:19:24,421 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:19:40,192 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:21:10,713 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:22:00,792 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:23:26,899 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:23:51,448 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:24:07,634 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:24:29,427 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:24:57,619 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:25:33,222 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:26:24,113 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:30:55,078 - distributed.utils_perf - WARNING - full garbage collections took 22% CPU time recently (threshold: 10%)
2022-11-17 14:31:05,034 - distributed.nanny - ERROR - Worker process died unexpectedly
2022-11-17 14:31:05,034 - distributed.nanny - ERROR - Worker process died unexpectedly
2022-11-17 14:31:05,035 - distributed.nanny - ERROR - Worker process died unexpectedly
2022-11-17 14:32:53,311 - distributed.worker - ERROR - Timed out during handshake while connecting to tcp://127.0.0.1:43286 after 30 s
Traceback (most recent call last):
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/distributed/comm/tcp.py", line 223, in read
    frames_nbytes = await stream.read_bytes(fmt_size)
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/asyncio/tasks.py", line 492, in wait_for
    fut.result()
asyncio.exceptions.CancelledError

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/site-packages/distributed/comm/core.py", line 328, in connect
    handshake = await asyncio.wait_for(comm.read(), time_left())
  File "/home/gertwk/.conda/envs/swyft32/lib/python3.9/asyncio/tasks.py", line 494, in wait_for
    raise exceptions.TimeoutError() from exc
asyncio.exceptions.TimeoutError

The last two errors then repeat indefinitely. If I increase the number of jobs/workers in cluster.scale(), I get rid of the garbage collection warnings, but still get the "worker process died unexpectedly"-error.

Do you know any likely reasons for these errors? I get the impression that they may be related to some memory overflow for the workers, so I have tried decreasing the batch_size parameter and increasing the number of processes, as well as increasing the memory parameter, but to no effect.

I'm not sure it matters, but I'll also mention that I am not passing the cluster to swyft.simulator.set_dask_cluster() as opposed to the example notebook (but this doesn't seem to be an issue, at least for lower numbers of simulations - the simulations are clearly still being processed in parallel). The reason I am not doing this is that I am defining the swyft.DaskSimulator in a separate class method, and passing the cluster to the latter seems to prevent the simulations from running. I am defining the cluster under the if __name__ = "__main__" statement though.

[fnattino]

Great that you are able to setup the cluster @GertKl! But unfortunately if the cluster is not passed to the simulator via swyft.simulator.set_dask_cluster(), SWYFT does not know how to reach it and it actually starts a new LocalCluster to run the simulations. So even though you are properly instantiating a PBSCluster, what is likely happening is that the simulations are run locally (in parallel, on the cores that you have available on the node where you are running the main script).

I am not sure why passing the cluster to the simulator, even if this is done in a separate class method, leads to problems. When you instantiate the PBSCluster and run cluster.scale() do you actually see jobs submitted to the Torque scheduler, e.g. via qstat? You should see one job per worker requested. Do the submitted jobs actually start (and keep) running? The PBSCluster does not have workers until the jobs start running, so if the queue in the cluster is particularly busy it might take a while before you can actually run simulations (so maybe this is what you have experienced?).

You could try to run the following snippet to verify whether the cluster is actually working properly:

import dask.array as da 
from dask.distributed import Client
from dask_jobqueue import PBSCluster

# configure your cluster
cluster = PBSCluster(...)

# connect to the cluster 
client = Client(cluster)

# run some random computation
x = da.random.random((10_000, 10_000), chunks=(2_000, 2_000))
y = da.dot(x, x.T).mean()
print(y.compute())  # this should block until completed, and print the result to stdout

I am also happy to have a look at the script to setup the simulator and the cluster if you want/can share it here, maybe in a sketched/simplified version.

[GertKl]

Hi again @fnattino,

first of all, I hope it is not a problem on your side that I have been slow at responding. My current work situation does not always allow me to do tests and follow up very quickly.

To answer your question: yes, when I run qstat, I can indeed see the jobs that have been submitted, and they do run (according to the output from qstat). However, this is not reflected in the output from top, which indicates that little to no CPU power is being used; the simulations do finish, but after way too much time. That is the case when I pass the cluster to my class method. When I don't, the CPU power seems to be about 100% times processes (the number of processes per job), as expected, though this seems to be independent of the number of jobs that I submit; is this expected, or should the CPU power scale with the number of jobs? In any case, the time spent on simulating then scales according to processes.

Concerning your code snippet, it does run, but extremely slowly (~10 minutes). Here too, top indicates little to no CPU activity.

I have discovered that printing cluster information by the statement print(cluster) yields something along the lines of

PBSCluster(76246660, 'tcp://129.177.41.21:39530', workers=0, threads=0, memory=0 B)

This is independent of where I put the statement (after initializing the cluster, after cluster.scale(), inside or outside the class method...). This is also the case for your code snippet, except when I use dask.distributed.LocalCluster instead (this also speeds up the calculation as expected). This seems to be a strong indication that the PBSCluster is not being initialized properly? Still, as mentioned, CPU power and time does scale with the processes input when not passing on the cluster.

I have written a toy version of my script, which reproduces the same problems, if you want to have a look at it. I've tried to make it so that it can be directly run without any changes (except maybe the queue input to PBSCluster).

import os
import numpy as np

x = np.arange(1,10,0.2)

def background(x, a=-2, b=3):
    return a*x + b

def sine(x, g=1, m=1, a=1, b=0.08):
    return (a-b*x)*np.sin(g*x + m)


def model(v):                                           
    y = background(x) + sine(x, g=v[0], m=v[1])    
    return dict(y=y)  

def noise(sim, v):                                     
    y = sim['y'] + np.random.randn(len(x))*1
    return dict(y=y)                                    



if __name__ == "__main__":
    
    params = [1,1]            #dummy observed parameter values
     
    from skeleton_swyft_analysis import SWYFT_analysis     #A class from a separate file,
                                                           #to compress and streamline the
                                                           #swyft analysis process

    S1 = SWYFT_analysis(model, noise, params)              #Initializing an object with which 
                                                           #to perform one entire SWYFT analysis.
                                                           #A.model and A.noise are methods of 
                                                           #the class ALP_sim, functioning as 
                                                           #model functions for the SWYFT analysis. 
    
    S1.param_names = ["Dummy1", "Dummy2"]                  #Parameter names         
    S1.nstore = 1000                                       #Number of simulations
    S1.path = os.getcwd() + "/dummy_directory"             #Path for directory to keep output
                                                           #and store.              
                    
    
    n_processes = 4                                        #Number of processes  
    n_jobs = 5                                             #Number of jobs

    from dask_jobqueue import PBSCluster

    cluster = PBSCluster(
        queue="short",
        memory="60GiB",  # max memory per job
        processes=n_processes,  # num processes per job
        cores=n_processes,  # num threads per job 
    # resource_spec="nodes=1:ppn=4",  # additionally configure request of resources  
        local_directory=S1.path,
        log_directory=S1.path,
        walltime="01:00:00",
    )
    
    # running the following print the job script which is submitted 
    # to the PBS scheduler to request one additional worker
    print(cluster.job_script())

    # add 5 workers to the Dask cluster => submit 5 jobs to the PBS scheduler
    cluster.scale(jobs=n_jobs) 
    
    S1.python_paths = [os.getcwd() + "/skeleton_swyft_analysis.py"]      #Paths to help workers find 
                                                                         #files to be imported. 
    
    S1.n_processes = n_processes                               #passes on n_processes to object,
                                                               #to automatically determine batch_size.        
    
    S1.run(cluster=None)                                       #Runs swyft analysis, while passing
                                                               #on the cluster to the class method.

In this form, the simulations speed up, and CPU activity increases, both in proportion with n_processes. Changing the last line to S1.run(cluster=cluster) passes the cluster to the class method and to swyft.simulator.set_dask_cluster(), and seems to suppress almost all CPU activity, thus slowing down simulations immensely.

The file skeleton_swyft_analysis.py:

import swyft

import numpy as np
import time

def process_time(s):
    h = int(s/3600)
    m = int((s-3600*h)/60)
    s = int(s-3600*h-60*m)
    return h, m, s


class SWYFT_analysis():
    
    def __init__(self, model, noise, params):
        
        self.model = model
        self.noise = noise
        
        self.params = params
        self.param_names = None
        
        self.prior = None
        self.simulator = None
        self.store = None
        
        self.prior_bounds = [[0,2],        #defines bounds of a uniform prior
                             [0,2]]
        
        
        self.output = self.model(self.params)                       #Creates a "True" observation
        self.output = self.noise(self.output, self.params)["y"]     #Adds noise to the observation
        self.output_shape = np.shape(self.output)        
        
        
        self.nstore = 1000
        
        self.n_processes = 4
        self.n_jobs = 1
        self.python_paths = None
        
        self.path = None
      
         
    
    def define_prior(self):
              
        lower = []
        upper = []
        
        for bounds in self.prior_bounds:
            
            lower.append(np.array(bounds[0]))
            upper.append(np.array(bounds[1]))
        
        
        self.prior = swyft.get_uniform_prior(lower, upper)
        
        
          
    def run(self, cluster=None):
        
        self.simulator = swyft.DaskSimulator(self.model, self.param_names, dict(y=self.output_shape) )

        self.simulator.set_dask_cluster(cluster)
                
        for path in self.python_paths:
                    
            self.simulator.client.upload_file(path)
                
            
        self.define_prior()

            
        self.store = swyft.Store.directory_store(self.path+"/store", simulator=self.simulator, overwrite=False)

        self.store.add(self.nstore, self.prior)
            
        print("Starting simulations")
        start=time.time()
        self.store.simulate(batch_size=int(np.ceil(self.store.__len__()/self.n_processes))) #batch_size = int(self.nstore/4 + 1) )
        stop = time.time()
        h,m,s = process_time(stop-start)
        print("Time elapsed during simulation: "+str(h)+" h, "+str(m)+" min, "+str(s)+" sec.")    

[fnattino]

Hi @GertKl

first of all, I hope it is not a problem on your side that I have been slow at responding. My current work situation does not always allow me to do tests and follow up very quickly.

No problem at all - I cannot guarantee quick answers myself, so feel free to take time!

To answer your question: yes, when I run qstat, I can indeed see the jobs that have been submitted, and they do run (according to the output from qstat). However, this is not reflected in the output from top, which indicates that little to no CPU power is being used; the simulations do finish, but after way too much time. That is the case when I pass the cluster to my class method.

OK, so when scaling the PBSCluster Dask correctly submits jobs to the scheduler and nodes are allocated to run the workers. Are you then running top on the same node where the main script is running? If so, it makes sense that you see no CPU usage, because the work is carried out by workers running on other compute nodes (the ones listed by qstat).

When I don't, the CPU power seems to be about 100% times processes (the number of processes per job), as expected, though this seems to be independent of the number of jobs that I submit; is this expected, or should the CPU power scale with the number of jobs? In any case, the time spent on simulating then scales according to processes.

When you don't pass the cluster to your class method, the PBSCluster that you set up is not used by Swyft. Swyft creates a Dask LocalCluster on the node where the main script is running and it uses this to run all the simulations. This is probably why you then see CPU activity on the node.
From your script below, I see that the value of n_processes determines the value of batch_size, and this is why, I think, you observed different timings.

Concerning your code snippet, it does run, but extremely slowly (~10 minutes). Here too, top indicates little to no CPU activity.

The computation there should be run via the PBSCluster - so again it makes sense you did not see CPU activity on the node where the script is running. However, 10 mins for running seems way too much. Is it possible that the queue was busy, and it took a lot of time for the worker jobs to actually start running? To double check you could try to run the following slightly modified snippet that measure the time of the actual computation. I have tried to run it myself on a cluster (changing PBS for SLURM, since we have a different scheduler) and I got a timing of ~10 sec in the stdout.

import dask.array as da
import time

from dask.distributed import Client
from dask_jobqueue import PBSCluster

# configure your cluster
n_processes = 4
n_jobs=5
cluster = PBSCluster(
  processes=n_processes,
  cores=n_processes,
  ...
)
cluster.scale(jobs=n_jobs)

# connect to the cluster 
client = Client(cluster)
client.wait_for_workers(n_jobs*n_processes)  # make sure all workers joined here

# run some random computation
x = da.random.random((10_000, 10_000), chunks=(2_000, 2_000))
y = da.dot(x, x.T).mean()

start = time.time()
print(y.compute())  # this should block until completed, and print the result to stdout
print(f"Elapsed: {time.time() - start}")

I have discovered that printing cluster information by the statement print(cluster) yields something along the lines of

PBSCluster(76246660, 'tcp://129.177.41.21:39530', workers=0, threads=0, memory=0 B)

This is independent of where I put the statement (after initializing the cluster, after cluster.scale(), inside or outside the class method...). This is also the case for your code snippet, except when I use dask.distributed.LocalCluster instead (this also speeds up the calculation as expected). This seems to be a strong indication that the PBSCluster is not being initialized properly? Still, as mentioned, CPU power and time does scale with the processes input when not passing on the cluster.

The fact that the PBSCluster shows workers=0 means that workers have not yet joined the cluster. When you call cluster.scale(...) worker jobs are submitted to the queue and only when these jobs start to run you will see the workers in the cluster. You can use the client's method client.wait_for_workers(n_workers) to block until n_workers have joined the cluster (also see snippet above and note that if you are asking for 4 processes and scaling to 5 jobs you are actually scaling to 5*4=20 workers). Printing the cluster after having waited for the workers should display something like:

PBSCluster(76246660, 'tcp://129.177.41.21:39530', workers=20, threads=20, memory=300GB)

I have written a toy version of my script, which reproduces the same problems, if you want to have a look at it. I've tried to make it so that it can be directly run without any changes (except maybe the queue input to PBSCluster).

Thanks for writing a simplified version of the script and for sharing it! I have run it on the cluster where I have access, and it seems to work. I noticed the following things:

1. You should really pass the cluster to your run method - you are otherwise not using the PBSCluster to run the simulations.
2. Right now batch_size is set as (number of simulations)/(number of cores in one job), but you probably want to set it as (number of simulations)/(number of cores in one job * number of jobs) so that all workers can be kept busy.
3. Especially now that you are testing, it might be useful to add a barrier to wait for all workers to become available before running the simulations. This way you are sure that your calculations actually reflect the usage of the desired number of workers. You could add in the skeleton_swyft_analysis:

...
self.simulator.set_dask_cluster(cluster)
self.simulator.client.wait_for_workers(n_processes*n_jobs)
...

4. For this toy example it is normal that calculations take longer when increasing the number of workers, because overhead dominates (for me, decreasing the number of jobs from 4 to 1 actually gave me a speed up). So I would not take timings of this toy example too seriously.

Last thing to notice: when scaling to a large number of simulations there is another element that might affect performance, probably significantly, and this is the store chunk size (=number of simulations stored in one file on disk). I would suggest to verify whether everything above works, after that you might want to try increasing the chunk size value and see whether this helps get better performances. To increase the chunk size you might need to change:

self.store = swyft.Store.directory_store(self.path+"/store", simulator=self.simulator, overwrite=False)

to:

import os
import zarr

self.store = swyft.Store(
  zarr_store=zarr.DirectoryStore(self.path+"/store"),
  simulator=self.simulator,
  chunksize=200,  # default is 1
  sync_path=self.path+"/store.sync",
  from_scratch=True,  # note this would overwrite the store
)

Apologies for the long answers, let me know if anything is unclear!

[GertKl]

Hi again,

Thank you for your long and detailed reply, I found it to be very helpful! A few things seem better now, but it's not quite working as I expect it to yet.

What I've found:

• Your test script seems to be running faster. Not quite ~10s, but ~120, which is considerably better than before. I didn't really make any changes though, so I'm not sure why. As to your hypothesis that it might be related to the queue being busy, it wasn't while it was clearly busy when I tried this time, so that shouldn't be the reason. Maybe I had just made some mistake before, it's a bit hard to know, but I'll take the improvement.
• Using the client.wait_for_workers(n_workers) line does indeed result in the expected output from print(cluster)
• I've confirmed that the work is indeed being done on a different node from the one where I run my script, at least today, when the node I run from is quite busy. I'm not sure why this would have happened when the other node wasn't busy (which I can see both from qstat and the cluster activity report), but it does explain why there was no CPU activity earlier. However, the simulations are still taking an unreasonable amount of time, according to $top, only about 2-3% CPU power is being used per process. This fluctuates up to ~30% from time to time, but I would expect it to be more or less constantly at ~100%, as it was when I apparently used the LocalCluster(). This problem occurs both for the toy script and the "real" script (although the toy still finishes in a reasonable amount of time). Do you have any explanations for this? I've tried increasing the available memory per job, which makes the CPU activity fluctuate more often, but it's overall the same problem (and I can't increase it more, without starting to use swap memory). I'm wondering if this could be indicative of a bottleneck resulting from the sending of data, back and forth between the node where the work is happening, and where I run the script. Is there a way to force the scheduler to choose the same node where I run the script? I haven't found one, so far.
• The chunk size tip was really helpful, at least it made a huge difference for the toy (for the "real" case it's impossible to say right now). I suppose the reason why you can't increase the chunk size to however high you want it is because the workers can only handle a limited amount of memory?

[fnattino]

Hi @GertKl - sorry for the late reply.

Happy to hear things are working better. In reply to some of your points above:

• If I understand correctly, you are running your main script in a job script that uses only part of a node of your cluster, and you would like to try forcing workers allocated via PBS to run on the same physical node, correct? Not sure whether you can ask PBS to allocate resources on a specific node. However, you could try the following: run your main script in a job that claims a full node, then use a LocalCluster to create workers, so you are sure they will run on the same node. You would only need to replace cluster = PBSCluster(...) with cluster = LocalCluster(n_workers=X, threads_per_worker=1) (with X = number of cores in a node), and keep simulator.set_dask_cluster(cluster). In this way you would test whether the communication between nodes is the real bottleneck in your experiments above.
• Concerning the store chunk size: its value represents the number of simulation results written in a single file. A good setting for this could actually be the batch_size, so that you avoid the situation of workers having to wait on each others to be able to write to the same file, but you still keep the number of files limited. The default chunk size (1) bypasses the issue of workers having to synchronise before writing to disk, but it can also lead to a large number of files in the storage - and, in turns, large overhead, as you have also observed.

[GertKl]

Hi again @fnattino, I hope you're still around!

I didn't make any significant progress with the Torque-cluster, unfortunately. However, since last time, I have gotten access to a SLURM cluster, which I am hoping could solve the issues, since now at least I know for sure that dask is meant to work on it. So far, I have verified at least that your code snippet from the comment on Nov 18th is running within the expected ~10 seconds. However, running my toy analysis (from Dec 1st) results in an error that I find mysterious, and which is hard to google. The error I get reads

Traceback (most recent call last):
  File "/fp/homes01/u01/ec-gertwk/CTA-Norway/ALP-project/Machine_learning_approach/skeleton/skeleton_script.py", line 85, in <module>
    S1.run(cluster=cluster)                                       #Runs swyft analysis, while passing
    ^^^^^^^^^^^^^^^^^^^^^^^
  File "/fp/homes01/u01/ec-gertwk/CTA-Norway/ALP-project/Machine_learning_approach/skeleton/skeleton_swyft_analysis.py", line 87, in run
    self.simulator.client.upload_file(path)
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/client.py", line 3617, in upload_file
    return self.register_worker_plugin(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/client.py", line 4781, in register_worker_plugin
    return self.sync(
           ^^^^^^^^^^
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/utils.py", line 338, in sync
    return sync(
           ^^^^^
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/utils.py", line 405, in sync
    raise exc.with_traceback(tb)
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/utils.py", line 378, in f
    result = yield future
             ^^^^^^^^^^^^
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/tornado/gen.py", line 769, in run
    value = future.result()
            ^^^^^^^^^^^^^^^
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/client.py", line 4707, in _register_worker_plugin
    raise exc.with_traceback(tb)
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/diagnostics/plugin.py", line 522, in setup
    response = await worker.upload_file(
  ^^^^^^^^^^^^^^^^^
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/worker.py", line 1280, in upload_file
    data = func(data)
  File "/fp/homes01/u01/ec-gertwk/.conda/envs/swyft32t/lib/python3.11/site-packages/distributed/worker.py", line 1273, in func
    with open(out_filename, "wb") as f:
^^^^^^^^^^^^^^^
FileNotFoundError: [Errno 2] No such file or directory: '/fp/homes01/u01/ec-gertwk/CTA-Norway/ALP-project/Machine_learning_approach/skeleton/output/dask-worker-space/worker-cp5z8jg1/skeleton_swyft_analysis.py'

The directory .../dask-worker-space/ does exist, and contains (after running the code) several folders with names like /worker-xxxxxxxx/, but not /worker-cp5z8jg1/. For now, I have almost no clue as to why the code is expecting this folder, when it doesn't seem to have been created. Does you have any ideas? Could it be related to some error in synchronization between the different nodes that are involved in the computations?

[fnattino]

Hello @GertKl!

Weird error indeed. Is it possible that the path to the local worker directory is not properly set, and it points to a path that is not accessible from some of the worker nodes? I see above that you were setting the local_directory argument using the cwd taken from the node where the main script runs, but maybe worker nodes on your system mount the home directory at different places (e.g. /fp/homes01/02/... etc.).

If possible I would thus try to set the local_directory argument without specifying a full path, but by using e.g. an environment variable. Often HPC systems use $TMPDIR or $SCRATCH to point to directories that are local on the workers, so trying something like local_directory="$TMPDIR" or local_directory="$SCRATCH" would also avoid any potential issues with the synchronisation of shared filesystems.

[GertKl]

Indeed, using ``$SCRATCH´´ did the trick :) With that, and several other tweaks, it now seems that the simulations are running as they should; what a relief! However, I haven't tried more than a few test runs so far, so it remains to be seen if nothing breaks down when I try making several orders of magnitude more simulations... But assuming that this works out fine, I would appreciate if you have any input on the following:

1. Do you have any rules of thumb as to when you should increase n_processes rather than n_jobs in order to speed up the simulations, or the other way around? And at what point should one stop increasing these parameters altogether? My understanding of what is happening under the hood is crude at best, and the internet hasn't given me any answers that are particularly clear on this. It seems crucial to understand for making something like 10⁶ or more simulations...
2. Obviously, SWYFT has already reached version 0.4, which @cweniger recommended to use rather than 0.3 (which I'm using now) at the beginning of this thread. Of course, this requires adapting to the associated interface overhaul, so I am not sure how much to prioritize this at the moment. What would you say are the main advantages of using 0.4? In particular, are there any major speed-ups?

[cweniger]

Regarding 2: Parallel simulations in v0.4 do not require dask anymore. Can simply instantiate the ZarrStore, and then run the simulations on several nodes simultaneously. The same file-blocking system that was already in v0.3 will prevent simulations overwriting each other. We use this now for all kinds of projects, and it seems to work pretty robustly without any extra setup problems.

The clear disadvantage w.r.t. the v0.3 dask implementation is that more advanced functionality, like the re-use of simulations from previous runs, or the simultaneous simulation over multiple different prior regions, is not possible right now (we plan to implement this again in the future). If you don't need this advanced functionality, you could in principle try v0.4. However, also the way simulators are implemented has completely changed (now this is a flexible graphical model), and this would require some extra work for migration. So, if you have something that works now, it is probably a good idea to stick to 0.3 for the time being. We will in the upcoming weeks/months improve the tutorials for v0.4, which should ease migration in the end.