feat(config): make number of threads configurable for Dask (#719)

Following the integration of Dask into REANA and discussions with
several REANA users, we identified the need to make the number of
threads configurable.

This commit adds the number of threads configuration to the `/api/info`
endpoint. It also validates that the requested number of threads for a
single Dask worker does not exceed the defined upper limit.

Closes reanahub/reana#874

Author: Alputer

docs/openapi.json

@@ -449,6 +449,10 @@
                   "title": "The amount of memory used by default by a single Dask worker",
                   "value": "2Gi"
                 },
+                "dask_cluster_default_single_worker_threads": {
+                  "title": "The number of threads used by default by a single Dask worker",
+                  "value": "4"
+                },
                 "dask_cluster_max_memory_limit": {
                   "title": "The maximum memory limit for Dask clusters created by users",
                   "value": "16Gi"
@@ -461,6 +465,10 @@
                   "title": "The maximum amount of memory that users can ask for the single Dask worker",
                   "value": "8Gi"
                 },
+                "dask_cluster_max_single_worker_threads": {
+                  "title": "The maximum number of threads that users can ask for the single Dask worker",
+                  "value": "8"
+                },
                 "dask_enabled": {
                   "title": "Dask workflows allowed in the cluster",
                   "value": "False"
@@ -606,6 +614,17 @@
                   },
                   "type": "object"
                 },
+                "dask_cluster_default_single_worker_threads": {
+                  "properties": {
+                    "title": {
+                      "type": "string"
+                    },
+                    "value": {
+                      "type": "string"
+                    }
+                  },
+                  "type": "object"
+                },
                 "dask_cluster_max_memory_limit": {
                   "properties": {
                     "title": {
@@ -639,6 +658,17 @@
                   },
                   "type": "object"
                 },
+                "dask_cluster_max_single_worker_threads": {
+                  "properties": {
+                    "title": {
+                      "type": "string"
+                    },
+                    "value": {
+                      "type": "string"
+                    }
+                  },
+                  "type": "object"
+                },
                 "dask_enabled": {
                   "properties": {
                     "title": {

reana_server/config.py

@@ -59,10 +59,10 @@
 )
 
 DASK_ENABLED = strtobool(os.getenv("DASK_ENABLED", "true"))
-"""Whether Dask is enabled in the cluster or not"""
+"""Whether Dask is enabled in the cluster or not."""
 
 DASK_AUTOSCALER_ENABLED = os.getenv("DASK_AUTOSCALER_ENABLED", "true").lower() == "true"
-"""Whether Dask autoscaler is enabled in the cluster or not"""
+"""Whether Dask autoscaler is enabled in the cluster or not."""
 
 REANA_DASK_CLUSTER_MAX_MEMORY_LIMIT = os.getenv(
     "REANA_DASK_CLUSTER_MAX_MEMORY_LIMIT", "16Gi"
@@ -72,7 +72,7 @@
 REANA_DASK_CLUSTER_DEFAULT_NUMBER_OF_WORKERS = int(
     os.getenv("REANA_DASK_CLUSTER_DEFAULT_NUMBER_OF_WORKERS", 2)
 )
-"""Number of workers in Dask cluster by default """
+"""Number of workers in Dask cluster by default."""
 
 REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS = int(
     os.getenv("REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS", 20)
@@ -89,6 +89,16 @@
 )
 """Maximum memory for one Dask worker."""
 
+REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_THREADS = int(
+    os.getenv("REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_THREADS", 4)
+)
+"""Number of threads for one Dask worker by default."""
+
+REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_THREADS = int(
+    os.getenv("REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_THREADS", 8)
+)
+"""Maximum number of threads for one Dask worker."""
+
 REANA_KUBERNETES_JOBS_MEMORY_LIMIT = os.getenv("REANA_KUBERNETES_JOBS_MEMORY_LIMIT")
 """Maximum memory limit for user job containers for workflow complexity estimation."""
 

reana_server/rest/info.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2021, 2022, 2024 CERN.
+# Copyright (C) 2021, 2022, 2024, 2025 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -34,6 +34,8 @@
     REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
     REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_MEMORY,
     REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS,
+    REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_THREADS,
+    REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_THREADS,
 )
 from reana_server.decorators import signin_required
 
@@ -237,6 +239,13 @@ def info(user, **kwargs):  # noqa
                   value:
                     type: string
                 type: object
+              dask_cluster_default_single_worker_threads:
+                properties:
+                  title:
+                    type: string
+                  value:
+                    type: string
+                type: object
               dask_cluster_max_single_worker_memory:
                 properties:
                   title:
@@ -251,6 +260,13 @@ def info(user, **kwargs):  # noqa
                   value:
                     type: string
                 type: object
+              dask_cluster_max_single_worker_threads:
+                properties:
+                  title:
+                    type: string
+                  value:
+                    type: string
+                type: object
             type: object
           examples:
             application/json:
@@ -355,6 +371,10 @@ def info(user, **kwargs):  # noqa
                     "title": "The amount of memory used by default by a single Dask worker",
                     "value": "2Gi"
                 },
+                "dask_cluster_default_single_worker_threads": {
+                    "title": "The number of threads used by default by a single Dask worker",
+                    "value": "4"
+                },
                 "dask_cluster_max_single_worker_memory": {
                     "title": "The maximum amount of memory that users can ask for the single Dask worker",
                     "value": "8Gi"
@@ -363,6 +383,10 @@ def info(user, **kwargs):  # noqa
                     "title": "The maximum number of workers that users can ask for the single Dask cluster",
                     "value": "20"
                 },
+                "dask_cluster_max_single_worker_threads": {
+                    "title": "The maximum number of threads that users can ask for the single Dask worker",
+                    "value": "8"
+                },
               }
         500:
           description: >-
@@ -480,6 +504,14 @@ def info(user, **kwargs):  # noqa
                 title="The maximum number of workers that users can ask for the single Dask cluster",
                 value=REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS,
             )
+            cluster_information["dask_cluster_default_single_worker_threads"] = dict(
+                title="The number of threads used by default by a single Dask worker",
+                value=REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_THREADS,
+            )
+            cluster_information["dask_cluster_max_single_worker_threads"] = dict(
+                title="The maximum number of threads that users can ask for the single Dask worker",
+                value=REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_THREADS,
+            )
 
         return InfoSchema().dump(cluster_information)
 
@@ -541,3 +573,5 @@ class InfoSchema(Schema):
         dask_cluster_default_single_worker_memory = fields.Nested(StringInfoValue)
         dask_cluster_max_single_worker_memory = fields.Nested(StringInfoValue)
         dask_cluster_max_number_of_workers = fields.Nested(StringInfoValue)
+        dask_cluster_default_single_worker_threads = fields.Nested(StringInfoValue)
+        dask_cluster_max_single_worker_threads = fields.Nested(StringInfoValue)

reana_server/rest/workflows.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2018, 2019, 2020, 2021, 2022, 2023, 2024 CERN.
+# Copyright (C) 2018, 2019, 2020, 2021, 2022, 2023, 2024, 2025 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -49,7 +49,7 @@
     validate_inputs,
     validate_workflow,
     validate_workspace_path,
-    validate_dask_memory_and_cores_limits,
+    validate_dask_limits,
 )
 from webargs import fields, validate
 from webargs.flaskparser import use_kwargs
@@ -567,7 +567,7 @@ def create_workflow(user):  # noqa
 
         validate_inputs(reana_spec_file)
 
-        validate_dask_memory_and_cores_limits(reana_spec_file)
+        validate_dask_limits(reana_spec_file)
 
         retention_days = reana_spec_file.get("workspace", {}).get("retention_days")
         retention_rules = get_workspace_retention_rules(retention_days)

reana_server/validation.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2022, 2024 CERN.
+# Copyright (C) 2022, 2024, 2025 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -29,6 +29,8 @@
     REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS,
     REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
     REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_MEMORY,
+    REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_THREADS,
+    REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_THREADS,
 )
 from reana_server import utils
 
@@ -165,7 +167,7 @@ def validate_retention_rule(rule: str, days: int) -> None:
         )
 
 
-def validate_dask_memory_and_cores_limits(reana_yaml: Dict) -> None:
+def validate_dask_limits(reana_yaml: Dict) -> None:
     """Validate Dask workflows are allowed in the cluster and memory limits are respected."""
     # Validate Dask workflows are allowed in the cluster
     dask_resources = reana_yaml["workflow"].get("resources", {}).get("dask", {})
@@ -195,6 +197,16 @@ def validate_dask_memory_and_cores_limits(reana_yaml: Dict) -> None:
                 f"The number of requested Dask workers ({number_of_workers}) exceeds the maximum limit ({REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS})."
             )
 
+        single_worker_threads = dask_resources.get(
+            "single_worker_threads",
+            REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_THREADS,
+        )
+
+        if single_worker_threads > REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_THREADS:
+            raise REANAValidationError(
+                f'The "single_worker_threads" provided in the dask resources exceeds the limit ({REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_THREADS}).'
+            )
+
         requested_dask_cluster_memory = (
             kubernetes_memory_to_bytes(single_worker_memory) * number_of_workers
         )

setup.py

@@ -48,7 +48,7 @@
     "Flask>=2.1.1,<2.3.0",  # same upper pin as invenio-base
     "gitpython>=3.1",
     "marshmallow>2.13.0,<3.0.0",
-    "reana-commons[kubernetes,yadage,snakemake,cwl]>=0.95.0a7,<0.96.0",
+    "reana-commons[kubernetes,yadage,snakemake,cwl]>=0.95.0a9,<0.96.0",
     "reana-db>=0.95.0a5,<0.96.0",
     "requests>=2.25.0",
     "tablib>=0.12.1",