use thread parameter in local cluster

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

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,

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

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,

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,