FEAT-#6914: Add a config for setting a number of threads per Dask wor…

…ker (#6915)

Signed-off-by: Igoshev, Iaroslav <iaroslav.igoshev@intel.com>

modin/config/__init__.py

@@ -21,6 +21,7 @@
     CIAWSAccessKeyID,
     CIAWSSecretAccessKey,
     CpuCount,
+    DaskThreadsPerWorker,
     DoUseCalcite,
     Engine,
     EnvironmentVariable,
@@ -73,6 +74,8 @@
     "RayRedisPassword",
     "TestRayClient",
     "LazyExecution",
+    # Dask specific
+    "DaskThreadsPerWorker",
     # Partitioning
     "NPartitions",
     "MinPartitionSize",

modin/config/envvars.py

@@ -830,6 +830,13 @@ class LazyExecution(EnvironmentVariable, type=bool):
     default = False
 
 
+class DaskThreadsPerWorker(EnvironmentVariable, type=int):
+    """Number of threads per Dask worker."""
+
+    varname = "MODIN_DASK_THREADS_PER_WORKER"
+    default = None
+
+
 def _check_vars() -> None:
     """
     Check validity of environment variables.

modin/core/execution/dask/common/utils.py

@@ -19,6 +19,7 @@
     CIAWSAccessKeyID,
     CIAWSSecretAccessKey,
     CpuCount,
+    DaskThreadsPerWorker,
     GithubCI,
     Memory,
     NPartitions,
@@ -44,12 +45,17 @@ def _disable_warnings():
         from distributed import Client
 
         num_cpus = CpuCount.get()
+        threads_per_worker = DaskThreadsPerWorker.get()
         memory_limit = Memory.get()
         worker_memory_limit = memory_limit // num_cpus if memory_limit else "auto"
 
         # when the client is initialized, environment variables are inherited
         with set_env(PYTHONWARNINGS="ignore::FutureWarning"):
-            client = Client(n_workers=num_cpus, memory_limit=worker_memory_limit)
+            client = Client(
+                n_workers=num_cpus,
+                threads_per_worker=threads_per_worker,
+                memory_limit=worker_memory_limit,
+            )
 
         if GithubCI.get():
             # set these keys to run tests that write to the mock s3 service. this seems