diff --git a/docs/parallel_config.md b/docs/parallel_config.md index dc0df3894f..719322351d 100644 --- a/docs/parallel_config.md +++ b/docs/parallel_config.md @@ -172,6 +172,9 @@ parameters: generally use 1 CPU per image analysis workflow, this is effectively the maximum number of concurrently running workflows. +* **threads_per_worker**: (int, optional, default = 1): the number of threads to run on each worker. In general this should +be left as 1 so that `n_workers` will let you accurately control the number of cores used. + * **memory**: (str, required, default = "1GB"): the amount of memory/RAM used per workflow. Can be set as a number plus units (KB, MB, GB, etc.). @@ -188,8 +191,8 @@ environmental variable. of key-value pairs (e.g. `{"getenv": "true"}`). !!! note - `n_workers` is the only parameter used by `LocalCluster`, all others are currently ignored. `n_workers`, - `memory`, and `disk` are required by the other clusters. All other parameters are optional. Additional parameters + `n_workers` and `threads_per_worker` are the only parameters used by `LocalCluster`, all others are currently ignored. + `n_workers`, `memory`, and `disk` are required by the other clusters. All other parameters are optional. Additional parameters defined in the [dask-jobqueue API](https://jobqueue.dask.org/en/latest/api.html) can be supplied. !!! note diff --git a/plantcv/parallel/jupyterconfig.py b/plantcv/parallel/jupyterconfig.py index 9032ebe31c..dc9254502c 100644 --- a/plantcv/parallel/jupyterconfig.py +++ b/plantcv/parallel/jupyterconfig.py @@ -42,6 +42,7 @@ def __init__(self): object.__setattr__(self, "cluster", "LocalCluster") object.__setattr__(self, "cluster_config", { "n_workers": 1, + "threads_per_worker": 1, "memory": "1GB", "disk": "1GB", "log_directory": None, diff --git a/plantcv/parallel/multiprocess.py b/plantcv/parallel/multiprocess.py index 365684dcb4..9431b61b8e 100644 --- a/plantcv/parallel/multiprocess.py +++ b/plantcv/parallel/multiprocess.py @@ -33,7 +33,8 @@ def create_dask_cluster(cluster, cluster_config): # If the requested cluster is a LocalCluster we get it from dask.distributed if cluster == "LocalCluster": # Create a local cluster client with n_workers - client = Client(n_workers=cluster_config.get("n_workers")) + client = Client(n_workers=cluster_config.get("n_workers"), + threads_per_worker=cluster_config.get("threads_per_worker", 1)) # Otherwise the cluster is a class from dask_jobqueue (a distributed resource scheduler) else: # if "cores" is not a key in the cluster_config then set it to 1 diff --git a/plantcv/parallel/workflowconfig.py b/plantcv/parallel/workflowconfig.py index 260b67d91b..8891aae679 100644 --- a/plantcv/parallel/workflowconfig.py +++ b/plantcv/parallel/workflowconfig.py @@ -33,6 +33,7 @@ def __init__(self): object.__setattr__(self, "cluster", "LocalCluster") object.__setattr__(self, "cluster_config", { "n_workers": 1, + "threads_per_worker": 1, "memory": "1GB", "disk": "1GB", "log_directory": None, diff --git a/tests/testdata/workflowconfig_template.json b/tests/testdata/workflowconfig_template.json index 9ad775bb58..8a45c83c1a 100644 --- a/tests/testdata/workflowconfig_template.json +++ b/tests/testdata/workflowconfig_template.json @@ -24,6 +24,7 @@ "cluster": "LocalCluster", "cluster_config": { "n_workers": 1, + "threads_per_worker": 1, "memory": "1GB", "disk": "1GB", "log_directory": null,