Speed up

distributed/scheduler.py

@@ -4773,16 +4773,19 @@ def _generate_taskstates(
             # run_spec in the submitted graph may be None. This happens
             # when an already persisted future is part of the graph
             elif k in dsk:
-                with dask.config.set({"tokenize.ensure-deterministic": True}):
-                    # The tokenize output _could_ be stored as an attribute if this
-                    # becomes too slow, but it's nice if we don't need to store this for
-                    # every task
-                    try:
-                        tok_lhs = tokenize(ts.run_spec)
-                        tok_rhs = tokenize(dsk[k])
-                    except RuntimeError:
-                        # Non-deterministic token
-                        tok_lhs = tok_rhs = None
+                # The tokenize output _could_ be stored as an attribute if this
+                # becomes too slow, but it's nice if we don't need to store this for
+                # every task
+                tok_lhs = tokenize(ts.run_spec)
+                tok_rhs = tokenize(dsk[k])
+                if tok_lhs != tok_rhs:
+                    # FIXME this is prone to threaded race conditions
+                    with dask.config.set({"tokenize.ensure-deterministic": True}):
+                        try:
+                            tokenize(ts.run_spec, dsk[k])
+                        except RuntimeError:
+                            # Non-deterministic token
+                            tok_lhs = tok_rhs = None
 
                 # Additionally check dependency names. This should only be necessary
                 # if run_specs can't be tokenized deterministically.