added a setting

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/core/settings.py

@@ -59,6 +59,12 @@ class ComputationalBackendSettings(BaseCustomSettings):
         ),
     ] = 50
     COMPUTATIONAL_BACKEND_DASK_CLIENT_ENABLED: bool = True
+    COMPUTATIONAL_BACKEND_DASK_CLIENT_MAX_DISTRIBUTED_CONCURRENCY: Annotated[
+        PositiveInt,
+        Field(
+            description="defines how many concurrent connections to each dask scheduler are allowed accross all director-v2 replicas"
+        ),
+    ] = 20
     COMPUTATIONAL_BACKEND_DEFAULT_CLUSTER_URL: Annotated[
         AnyUrl,
         Field(

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_dask.py

@@ -70,7 +70,6 @@
 _DASK_CLIENT_RUN_REF: Final[str] = "{user_id}:{project_id}:{run_id}"
 _TASK_RETRIEVAL_ERROR_TYPE: Final[str] = "task-result-retrieval-timeout"
 _TASK_RETRIEVAL_ERROR_CONTEXT_TIME_KEY: Final[str] = "check_time"
-_DASK_CLUSTER_CLIENT_SEMAPHORE_CAPACITY: Final[int] = 20
 
 
 def _get_redis_client_from_scheduler(
@@ -90,10 +89,20 @@ def _get_semaphore_cluster_redis_key(
     return f"{APP_NAME}-cluster-user_id_{user_id}-wallet_id_{run_metadata.get('wallet_id')}"
 
 
+def _get_semaphore_capacity_from_scheduler(
+    _user_id: UserID,
+    scheduler: "DaskScheduler",
+    **kwargs,  # pylint: disable=unused-argument # noqa: ARG001
+) -> int:
+    return (
+        scheduler.settings.COMPUTATIONAL_BACKEND_DASK_CLIENT_MAX_DISTRIBUTED_CONCURRENCY
+    )
+
+
 @with_limited_concurrency_cm(
     _get_redis_client_from_scheduler,
     key=_get_semaphore_cluster_redis_key,
-    capacity=_DASK_CLUSTER_CLIENT_SEMAPHORE_CAPACITY,
+    capacity=_get_semaphore_capacity_from_scheduler,
     blocking=True,
     blocking_timeout=None,
 )