Add config option for direct-to-workers (#9097)

Author: jrbourbeau

distributed/client.py

@@ -1084,6 +1084,8 @@ def __init__(
         if deserializers is None:
             deserializers = serializers
         self._deserializers = deserializers
+        if direct_to_workers is None:
+            direct_to_workers = dask.config.get("distributed.client.direct-to-workers")
         self.direct_to_workers = direct_to_workers
         self._previous_as_current = None
 

distributed/distributed-schema.yaml

@@ -678,6 +678,11 @@ properties:
           Configuration settings for Dask Clients
 
         properties:
+          direct-to-workers:
+            type: [boolean, 'null']
+            description:
+              Whether to connect directly to workers for gather / scatter.
+
           heartbeat:
             type: string
             description:

distributed/distributed.yaml

@@ -201,6 +201,7 @@ distributed:
       OPENBLAS_NUM_THREADS: 1
 
   client:
+    direct-to-workers: null # Whether to connect directly to workers for gather / scatter
     heartbeat: 5s  # Interval between client heartbeats
     scheduler-info-interval: 2s  # Interval between scheduler-info updates
     security-loader: null  # A callable to load security credentials if none are provided explicitl

distributed/tests/test_client.py

@@ -6002,12 +6002,22 @@ def test_no_threads_lingering():
 
 @gen_cluster()
 async def test_direct_async(s, a, b):
+    # Keyword option
     async with Client(s.address, asynchronous=True, direct_to_workers=True) as c:
         assert c.direct_to_workers
 
     async with Client(s.address, asynchronous=True, direct_to_workers=False) as c:
         assert not c.direct_to_workers
 
+    # Config option
+    with dask.config.set({"distributed.client.direct-to-workers": True}):
+        async with Client(s.address, asynchronous=True) as c:
+            assert c.direct_to_workers
+
+    with dask.config.set({"distributed.client.direct-to-workers": False}):
+        async with Client(s.address, asynchronous=True) as c:
+            assert not c.direct_to_workers
+
 
 def test_direct_sync(c):
     assert not c.direct_to_workers