Update inlining Futures in task graph in Client._graph_to_futures

distributed/tests/test_utils_comm.py

@@ -23,6 +23,11 @@ def test_subs_multiple():
     dsk = {"a": (sum, ["x", "y"])}
     assert subs_multiple(dsk, data) == {"a": (sum, [1, 2])}
 
+    # Tuple key
+    data = {"x": 1, ("y", 0): 2}
+    dsk = {"a": (sum, ["x", ("y", 0)])}
+    assert subs_multiple(dsk, data) == {"a": (sum, [1, 2])}
+
 
 @gen_cluster(client=True)
 def test_gather_from_workers_permissive(c, s, a, b):

distributed/utils_comm.py

@@ -15,6 +15,8 @@
 
 logger = logging.getLogger(__name__)
 
+__no_value__ = "__no_value__"
+
 
 async def gather_from_workers(who_has, rpc, close=True, serializers=None, who=None):
     """ Gather data directly from peers
@@ -299,15 +301,20 @@ def subs_multiple(o, d):
 
     """
     typ = type(o)
-    if typ is tuple and o and callable(o[0]):  # istask(o)
+    if not (typ is tuple and o and callable(o[0])):  # not istask(o)
+        try:  # Is o a key to substitute?
+            result = d.get(o, __no_value__)
+            if result is not __no_value__:
+                return result
+        except TypeError:  # unhashable type
+            pass
+        if typ is list:
+            return typ([subs_multiple(i, d) for i in o])
+        elif typ is dict:
+            return {k: subs_multiple(v, d) for (k, v) in o.items()}
+        return o
+    else:
         return (o[0],) + tuple(subs_multiple(i, d) for i in o[1:])
-    elif typ is list:
-        return typ([subs_multiple(i, d) for i in o])
-    elif typ is dict:
-        return {k: subs_multiple(v, d) for (k, v) in o.items()}
-    elif typ is str:
-        return d.get(o, o)
-    return o
 
 
 retry_count = dask.config.get("distributed.comm.retry.count")