eriknw: Looking now.
eriknw: Great.  I think the tradeoff is worth it for these functions.  Not passing an optional `dependencies` to non-critical functions also seems reasonable as a way to avoid complications.

A downside is that now I don't know the next "obvious" thing to do to optimize these further!  What do profiles look like with these changes?
mrocklin: I recommend pulling out snakeviz and running the following:

```python
In [1]: import dask.array as da
In [2]: n = 12
In [3]: x = da.random.random((2**n, 2**n), chunks=(2**n, 1))   # 4096 x 4096, 4096 x 1
In [4]: for i in range(1, n):                                  # 4096 x 4096, 2048 x 2
            x = x.rechunk((2**(n - i), 2**(i + 1)))            # 4096 x 4096, 1024 x 4
   ...:     
In [5]: y = x.sum()
In [6]: %load_ext snakeviz
In [7]: %snakeviz y._optimize(y.dask, y._keys())
```

Inline_functions takes up about half of the optimization time now.  Really though, I think we're at the point of diminishing returns.  This computation takes around a minute-plus to run on a distributed machine.


mrocklin: If you are looking to do some optimization work we're close to the point where we might want to Cythonize a couple of functions within the distributed scheduler.  If you're interested let me know and I'll produce a benchmark script for you to play with.
eriknw: Okay.  I'm up for Cythonizing some functions.  I'm taking a vacation in a few days, so it'll probably be a few weeks until I can do the work.  Thanks for letting me know this would be beneficial to do.
diff --git a/dask/array/optimization.py b/dask/array/optimization.py
index 8d7224fd4..6a201ead5 100644
--- a/dask/array/optimization.py
+++ b/dask/array/optimization.py
@@ -21,10 +21,11 @@ def optimize(dsk, keys, **kwargs):
     keys = list(flatten(keys))
     fast_functions = kwargs.get('fast_functions',
                              set([getarray, np.transpose]))
-    dsk2 = cull(dsk, keys)
-    dsk4 = fuse(dsk2, keys)
+    dsk2, dependencies = cull(dsk, keys)
+    dsk4, dependencies = fuse(dsk2, keys, dependencies)
     dsk5 = optimize_slices(dsk4)
-    dsk6 = inline_functions(dsk5, keys, fast_functions=fast_functions)
+    dsk6 = inline_functions(dsk5, keys, fast_functions=fast_functions,
+            dependencies=dependencies)
     return dsk6
 
 
diff --git a/dask/async.py b/dask/async.py
index 4f9d92b1c..2d8fbb87b 100644
--- a/dask/async.py
+++ b/dask/async.py
@@ -434,7 +434,7 @@ def get_async(apply_async, num_workers, dsk, result, cache=None,
     for f in start_cbs:
         f(dsk)
 
-    dsk = cull(dsk, list(results))
+    dsk, dependencies = cull(dsk, list(results))
 
     keyorder = order(dsk)
 
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 3348be39a..268a17931 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -64,7 +64,7 @@ def lazify(dsk):
     return valmap(lazify_task, dsk)
 
 
-def inline_singleton_lists(dsk):
+def inline_singleton_lists(dsk, dependencies=None):
     """ Inline lists that are only used once
 
     >>> d = {'b': (list, 'a'),
@@ -74,8 +74,8 @@ def inline_singleton_lists(dsk):
 
     Pairs nicely with lazify afterwards
     """
-
-    dependencies = dict((k, get_dependencies(dsk, k)) for k in dsk)
+    if dependencies is None:
+        dependencies = dict((k, get_dependencies(dsk, k)) for k in dsk)
     dependents = reverse_dict(dependencies)
 
     keys = [k for k, v in dsk.items()
@@ -85,9 +85,9 @@ def inline_singleton_lists(dsk):
 
 def optimize(dsk, keys, **kwargs):
     """ Optimize a dask from a dask.bag """
-    dsk2 = cull(dsk, keys)
-    dsk3 = fuse(dsk2, keys)
-    dsk4 = inline_singleton_lists(dsk3)
+    dsk2, dependencies = cull(dsk, keys)
+    dsk3, dependencies = fuse(dsk2, keys, dependencies)
+    dsk4 = inline_singleton_lists(dsk3, dependencies)
     dsk5 = lazify(dsk4)
     return dsk5
 
diff --git a/dask/core.py b/dask/core.py
index 275748384..ead939fd0 100644
--- a/dask/core.py
+++ b/dask/core.py
@@ -319,7 +319,7 @@ def subs(task, key, val):
     return task[:1] + tuple(newargs)
 
 
-def _toposort(dsk, keys=None, returncycle=False):
+def _toposort(dsk, keys=None, returncycle=False, dependencies=None):
     # Stack-based depth-first search traversal.  This is based on Tarjan's
     # method for topological sorting (see wikipedia for pseudocode)
     if keys is None:
@@ -340,6 +340,9 @@ def _toposort(dsk, keys=None, returncycle=False):
     # that has already been added to `completed`.
     seen = set()
 
+    if dependencies is None:
+        dependencies = dict((k, get_dependencies(dsk, k)) for k in dsk)
+
     for key in keys:
         if key in completed:
             continue
@@ -355,7 +358,7 @@ def _toposort(dsk, keys=None, returncycle=False):
 
             # Add direct descendants of cur to nodes stack
             next_nodes = []
-            for nxt in get_dependencies(dsk, cur):
+            for nxt in dependencies[cur]:
                 if nxt not in completed:
                     if nxt in seen:
                         # Cycle detected!
@@ -385,9 +388,9 @@ def _toposort(dsk, keys=None, returncycle=False):
     return ordered
 
 
-def toposort(dsk):
+def toposort(dsk, dependencies=None):
     """ Return a list of keys of dask sorted in topological order."""
-    return _toposort(dsk)
+    return _toposort(dsk, dependencies=dependencies)
 
 
 def getcycle(d, keys):
diff --git a/dask/dataframe/optimize.py b/dask/dataframe/optimize.py
index 0ae46d54c..d23a7ff2e 100644
--- a/dask/dataframe/optimize.py
+++ b/dask/dataframe/optimize.py
@@ -15,9 +15,9 @@ def fuse_castra_index(dsk):
 
 def optimize(dsk, keys, **kwargs):
     if isinstance(keys, list):
-        dsk2 = cull(dsk, list(core.flatten(keys)))
+        dsk2, dependencies = cull(dsk, list(core.flatten(keys)))
     else:
-        dsk2 = cull(dsk, [keys])
+        dsk2, dependencies = cull(dsk, [keys])
     try:
         from castra import Castra
         dsk3 = fuse_getitem(dsk2, Castra.load_partition, 3)
@@ -25,5 +25,5 @@ def optimize(dsk, keys, **kwargs):
     except ImportError:
         dsk4 = dsk2
     dsk5 = fuse_getitem(dsk4, dataframe_from_ctable, 3)
-    dsk6 = cull(dsk5, keys)
+    dsk6, _ = cull(dsk5, keys)
     return dsk6
diff --git a/dask/dataframe/shuffle.py b/dask/dataframe/shuffle.py
index 0b4c49ab0..fb797a503 100644
--- a/dask/dataframe/shuffle.py
+++ b/dask/dataframe/shuffle.py
@@ -144,7 +144,7 @@ def set_partition(df, index, divisions, compute=False, drop=True, **kwargs):
         dsk.update(index.dask)
 
     if compute:
-        dsk = cull(dsk, list(dsk4.keys()))
+        dsk, _ = cull(dsk, list(dsk4.keys()))
 
     return DataFrame(dsk, name, metadata, divisions)
 
diff --git a/dask/multiprocessing.py b/dask/multiprocessing.py
index dce045998..a2f6b996f 100644
--- a/dask/multiprocessing.py
+++ b/dask/multiprocessing.py
@@ -68,8 +68,9 @@ def get(dsk, keys, optimizations=[], num_workers=None,
                                           func_loads=func_loads)
 
     # Optimize Dask
-    dsk2 = fuse(dsk, keys)
-    dsk3 = pipe(dsk2, partial(cull, keys=keys), *optimizations)
+    dsk2, dependencies = cull(dsk, keys)
+    dsk3, dependencies = fuse(dsk2, keys, dependencies)
+    dsk4 = pipe(dsk3, *optimizations)
 
     try:
         # Run
diff --git a/dask/optimize.py b/dask/optimize.py
index e557ccd73..c34b6e4c7 100644
--- a/dask/optimize.py
+++ b/dask/optimize.py
@@ -22,49 +22,80 @@ def cull(dsk, keys):
     Examples
     --------
     >>> d = {'x': 1, 'y': (inc, 'x'), 'out': (add, 'x', 10)}
-    >>> cull(d, 'out')  # doctest: +SKIP
+    >>> dsk, dependencies = cull(d, 'out')  # doctest: +SKIP
+    >>> dsk  # doctest: +SKIP
     {'x': 1, 'out': (add, 'x', 10)}
+    >>> dependencies  # doctest: +SKIP
+    {'x': set(), 'out': set(['x'])}
+
+    Returns
+    -------
+    dsk: culled dask graph
+    dependencies: Dict mapping {key: [deps]}.  Useful side effect to accelerate
+        other optimizations, notably fuse.
     """
     if not isinstance(keys, (list, set)):
         keys = [keys]
-    nxt = set(flatten(keys))
-    seen = nxt
-    while nxt:
-        cur = nxt
-        nxt = set()
-        for item in cur:
-            for dep in get_dependencies(dsk, item):
-                if dep not in seen:
-                    nxt.add(dep)
-        seen.update(nxt)
-    return dict((k, v) for k, v in dsk.items() if k in seen)
-
-
-def fuse(dsk, keys=None):
-    """ Return new dask with linear sequence of tasks fused together.
+    out = dict()
+    seen = set()
+    dependencies = dict()
+    stack = list(set(flatten(keys)))
+    while stack:
+        key = stack.pop()
+        out[key] = dsk[key]
+        deps = get_dependencies(dsk, key, as_list=True)  # fuse needs lists
+        dependencies[key] = deps
+        unseen = [d for d in deps if d not in seen]
+        stack.extend(unseen)
+        seen.update(unseen)
+    return out, dependencies
+
+
+def fuse(dsk, keys=None, dependencies=None):
+    """ Return new dask graph with linear sequence of tasks fused together.
 
     If specified, the keys in ``keys`` keyword argument are *not* fused.
+    Supply ``dependencies`` from output of ``cull`` if available to avoid
+    recomputing dependencies.
 
-    This may be used as an optimization step.
+    Parameters
+    ----------
+    dsk: dict
+    keys: list
+    dependencies: dict, optional
+        {key: [list-of-keys]}.  Must be a list to provide count of each key
+        This optional input often comes from ``cull``
 
     Examples
     --------
     >>> d = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')}
-    >>> fuse(d)  # doctest: +SKIP
+    >>> dsk, dependencies = fuse(d)
+    >>> dsk # doctest: +SKIP
     {'c': (inc, (inc, 1))}
-    >>> fuse(d, keys=['b'])  # doctest: +SKIP
+    >>> dsk, dependencies = fuse(d, keys=['b'])
+    >>> dsk  # doctest: +SKIP
     {'b': (inc, 1), 'c': (inc, 'b')}
+
+    Returns
+    -------
+    dsk: output graph with keys fused
+    dependencies: dict mapping dependencies after fusion.  Useful side effect
+        to accelerate other downstream optimizations.
     """
     if keys is not None and not isinstance(keys, set):
         if not isinstance(keys, list):
             keys = [keys]
         keys = set(flatten(keys))
 
+    if dependencies is None:
+        dependencies = dict((key, get_dependencies(dsk, key, as_list=True))
+                            for key in dsk)
+
     # locate all members of linear chains
     child2parent = {}
     unfusible = set()
     for parent in dsk:
-        deps = get_dependencies(dsk, parent, as_list=True)
+        deps = dependencies[parent]
         has_many_children = len(deps) > 1
         for child in deps:
             if keys is not None and child in keys:
@@ -94,6 +125,8 @@ def fuse(dsk, keys=None):
             chain.append(child)
         chains.append(chain)
 
+    dependencies = dict((k, set(v)) for k, v in dependencies.items())
+
     # create a new dask with fused chains
     rv = {}
     fused = set()
@@ -102,6 +135,8 @@ def fuse(dsk, keys=None):
         val = dsk[child]
         while chain:
             parent = chain.pop()
+            dependencies[parent].update(dependencies.pop(child))
+            dependencies[parent].remove(child)
             val = subs(dsk[parent], child, val)
             fused.add(child)
             child = parent
@@ -111,10 +146,10 @@ def fuse(dsk, keys=None):
     for key, val in dsk.items():
         if key not in fused:
             rv[key] = val
-    return rv
+    return rv, dependencies
 
 
-def inline(dsk, keys=None, inline_constants=True):
+def inline(dsk, keys=None, inline_constants=True, dependencies=None):
     """ Return new dask with the given keys inlined with their values.
 
     Inlines all constants if ``inline_constants`` keyword is True.
@@ -140,13 +175,17 @@ def inline(dsk, keys=None, inline_constants=True):
     if inline_constants:
         keys.update(k for k, v in dsk.items() if not istask(v))
 
+    if dependencies is None:
+        dependencies = dict((k, get_dependencies(dsk, k)) for k in dsk)
+
     # Keys may depend on other keys, so determine replace order with toposort.
     # The values stored in `keysubs` do not include other keys.
-    replaceorder = toposort(dict((k, dsk[k]) for k in keys if k in dsk))
+    replaceorder = toposort(dict((k, dsk[k]) for k in keys if k in dsk),
+                            dependencies=dependencies)
     keysubs = {}
     for key in replaceorder:
         val = dsk[key]
-        for dep in keys & get_dependencies(dsk, key):
+        for dep in keys & dependencies[key]:
             if dep in keysubs:
                 replace = keysubs[dep]
             else:
@@ -159,13 +198,14 @@ def inline(dsk, keys=None, inline_constants=True):
     for key, val in dsk.items():
         if key in keys:
             continue
-        for item in keys & get_dependencies(dsk, key):
+        for item in keys & dependencies[key]:
             val = subs(val, item, keysubs[item])
         rv[key] = val
     return rv
 
 
-def inline_functions(dsk, output, fast_functions=None, inline_constants=False):
+def inline_functions(dsk, output, fast_functions=None, inline_constants=False,
+        dependencies=None):
     """ Inline cheap functions into larger operations
 
     Examples
@@ -194,7 +234,8 @@ def inline_functions(dsk, output, fast_functions=None, inline_constants=False):
 
     fast_functions = set(fast_functions)
 
-    dependencies = dict((k, get_dependencies(dsk, k)) for k in dsk)
+    if dependencies is None:
+        dependencies = dict((k, get_dependencies(dsk, k)) for k in dsk)
     dependents = reverse_dict(dependencies)
 
     keys = [k for k, v in dsk.items()
@@ -203,7 +244,8 @@ def inline_functions(dsk, output, fast_functions=None, inline_constants=False):
               and dependents[k]
               and k not in output]
     if keys:
-        return inline(dsk, keys, inline_constants=inline_constants)
+        return inline(dsk, keys, inline_constants=inline_constants,
+                dependencies=dependencies)
     else:
         return dsk
 
@@ -515,7 +557,7 @@ def fuse_selections(dsk, head1, head2, merge):
     ...        'y': (getitem, 'x', 'a')}
     >>> merge = lambda t1, t2: (load, t2[1], t2[2], t1[2])
     >>> dsk2 = fuse_selections(dsk, getitem, load, merge)
-    >>> cull(dsk2, 'y')
+    >>> cull(dsk2, 'y')[0]
     {'y': (<function load at ...>, 'store', 'part', 'a')}
     """
     dsk2 = dict()
@@ -548,7 +590,7 @@ def fuse_getitem(dsk, func, place):
     >>> dsk = {'x': (load, 'store', 'part', ['a', 'b']),
     ...        'y': (getitem, 'x', 'a')}
     >>> dsk2 = fuse_getitem(dsk, load, 3)  # columns in arg place 3
-    >>> cull(dsk2, 'y')
+    >>> cull(dsk2, 'y')[0]
     {'y': (<function load at ...>, 'store', 'part', 'a')}
     """
     return fuse_selections(dsk, getitem, func,
diff --git a/dask/order.py b/dask/order.py
index 60fd03787..06f602c5d 100644
--- a/dask/order.py
+++ b/dask/order.py
@@ -58,10 +58,10 @@ from __future__ import absolute_import, division, print_function
 
 from operator import add
 
-from .core import get_deps
+from .core import get_dependencies, reverse_dict, get_deps
 
 
-def order(dsk):
+def order(dsk, dependencies=None):
     """ Order nodes in dask graph
 
     The ordering will be a toposort but will also have other convenient
@@ -74,7 +74,10 @@ def order(dsk):
     >>> order(dsk)
     {'a': 2, 'c': 1, 'b': 3, 'd': 0}
     """
-    dependencies, dependents = get_deps(dsk)
+    if dependencies is None:
+        dependencies = dict((k, get_dependencies(dsk, k)) for k in dsk)
+    dependents = reverse_dict(dependencies)
+
     ndeps = ndependents(dependencies, dependents)
     maxes = child_max(dependencies, dependents, ndeps)
     return dfs(dependencies, dependents, key=maxes.get)

Tune dask.array.optimize
In some pathological (yet also real) cases, graph optimization can cost about as much as computation:

I've been benchmarking against this computation

```python
import dask.array as da
n = 12
x = da.random.random((2**n, 2**n), chunks=(2**n, 1))   # 4096 x 4096, 4096 x 1
for i in range(1, n):                                  # 4096 x 4096, 2048 x 2
        x = x.rechunk((2**(n - i), 2**(i + 1)))            # 4096 x 4096, 1024 x 4

y = x.sum()

>>> %time y.compute()
CPU times: user 39 s, sys: 2.45 s, total: 41.5 s
Wall time: 39.1 s

>>> %prun  y._optimize(y.dask, y._keys())
```

```

         29816630 function calls (28569410 primitive calls) in 22.570 seconds

   Ordered by: internal time

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
   604153    3.733    0.000    8.800    0.000 core.py:189(get_dependencies)
  6240188    2.504    0.000    2.843    0.000 core.py:27(istask)
  1312754    2.414    0.000    3.091    0.000 core.py:154(_deps)
        3    2.146    0.715    2.146    0.715 core.py:284(<listcomp>)
   299004    1.456    0.000    3.326    0.000 rewrite.py:375(_match)
909302/77823    1.331    0.000    2.017    0.000 optimize.py:211(functions_of)
  2009082    0.625    0.000    0.625    0.000 {built-in method isinstance}
        3    0.618    0.206    2.821    0.940 core.py:275(reverse_dict)
299004/77823    0.553    0.000    5.650    0.000 rewrite.py:363(_bottom_up)
   299004    0.480    0.000    4.206    0.000 rewrite.py:283(iter_matches)
   348156    0.426    0.000    0.426    0.000 rewrite.py:50(__init__)
   874487    0.406    0.000    1.014    0.000 rewrite.py:8(head)
  2303972    0.403    0.000    0.403    0.000 {method 'pop' of 'list' objects}
219136/51200    0.400    0.000    0.990    0.000 core.py:291(subs)
  2250724    0.391    0.000    0.391    0.000 {method 'extend' of 'list' objects}
        2    0.376    0.188    4.363    2.182 optimize.py:117(inline)
    49152    0.341    0.000    0.673    0.000 core.py:307(<listcomp>)
  4671438    0.339    0.000    0.339    0.000 {built-in method callable}
   299004    0.270    0.000    4.477    0.000 rewrite.py:304(_rewrite)
   794616    0.257    0.000    1.211    0.000 rewrite.py:81(current)
        1    0.204    0.204    1.261    1.261 optimize.py:43(fuse)
   745465    0.186    0.000    0.186    0.000 {method 'get' of 'dict' objects}
  1484786    0.168    0.000    0.168    0.000 {method 'append' of 'list' objects}
        2    0.133    0.066    0.752    0.376 core.py:321(_toposort)
        1    0.127    0.127    1.249    1.249 optimize.py:16(cull)
   348156    0.126    0.000    0.126    0.000 {method 'pop' of 'collections.deque' objects}
    79871    0.125    0.000    0.185    0.000 optimization.py:32(is_full_slice)
   745464    0.122    0.000    0.122    0.000 rewrite.py:119(edges)
        1    0.122    0.122    2.202    2.202 optimize.py:200(<listcomp>)
   595958    0.113    0.000    0.113    0.000 {method 'add' of 'set' objects}
        1    0.106    0.106    5.805    5.805 {method 'update' of 'dict' objects}
225278/221182    0.106    0.000    3.578    0.000 rewrite.py:365(<genexpr>)
        1    0.098    0.098    6.363    6.363 optimize.py:237(dealias)
   176126    0.093    0.000    0.175    0.000 rewrite.py:19(args)
    81919    0.086    0.000    0.086    0.000 {built-in method hasattr}
    49152    0.070    0.000    0.135    0.000 rewrite.py:70(next)
    79871    0.069    0.000    0.069    0.000 {built-in method hash}
47103/24575    0.067    0.000    1.377    0.000 rewrite.py:367(<listcomp>)
    79872    0.066    0.000    1.486    0.000 optimize.py:287(<genexpr>)
    79872    0.065    0.000    1.344    0.000 optimize.py:278(<genexpr>)
    79871    0.063    0.000    0.132    0.000 core.py:9(ishashable)
    77824    0.061    0.000    1.311    0.000 optimize.py:197(<genexpr>)
    49152    0.058    0.000    0.084    0.000 rewrite.py:62(copy)
        1    0.058    0.058    6.676    6.676 optimization.py:49(remove_full_slices)
        1    0.054    0.054   22.570   22.570 <string>:1(<module>)
    77823    0.049    0.000    5.699    0.000 rewrite.py:315(rewrite)
    79872    0.044    0.000    0.044    0.000 optimize.py:40(<genexpr>)
    28675    0.042    0.000    0.042    0.000 {method 'values' of 'dict' objects}
        1    0.039    0.039    7.491    7.491 optimize.py:168(inline_functions)
    98304    0.038    0.000    0.038    0.000 optimization.py:46(<genexpr>)
        1    0.037    0.037    0.169    0.169 optimize.py:281(<genexpr>)
        1    0.036    0.036    0.221    0.221 optimization.py:69(<genexpr>)
    79872    0.033    0.000    0.033    0.000 optimization.py:71(<genexpr>)
        1    0.033    0.033   22.515   22.515 optimization.py:14(optimize)
    49152    0.032    0.000    0.780    0.000 core.py:314(<listcomp>)
    79871    0.031    0.000    0.117    0.000 optimize.py:231(unwrap_partial)
    49154    0.026    0.000    0.026    0.000 optimize.py:145(<genexpr>)
    49160    0.018    0.000    0.028    0.000 core.py:246(flatten)
        2    0.011    0.006    0.011    0.006 optimize.py:282(<genexpr>)
    49152    0.011    0.000    0.011    0.000 {method 'extend' of 'collections.deque' objects}
        1    0.009    0.009    0.009    0.009 {method 'copy' of 'dict' objects}
    49152    0.009    0.000    0.037    0.000 {built-in method all}
    77823    0.008    0.000    0.008    0.000 {method 'issubset' of 'set' objects}
    79872    0.008    0.000    0.008    0.000 {built-in method len}
```

I also recommend looking at this with snakeviz

```python
%load_ext snakeviz
%snakeviz y._optimize(y.dask, y._keys())
```

diff --git a/dask/array/tests/test_optimization.py b/dask/array/tests/test_optimization.py
index 37d02d38d..0ff08fb2f 100644
--- a/dask/array/tests/test_optimization.py
+++ b/dask/array/tests/test_optimization.py
@@ -69,14 +69,14 @@ def test_optimize_slicing():
            'e': (getarray, 'd', (slice(None, None, None),))}
 
     expected = {'e': (getarray, (range, 10), (slice(0, 5, None),))}
-    result = optimize_slices(fuse(dsk, []))
+    result = optimize_slices(fuse(dsk, [])[0])
     assert result == expected
 
     # protect output keys
     expected = {'c': (range, 10),
                 'd': (getarray, 'c', (slice(0, 5, None),)),
                 'e': 'd'}
-    result = optimize_slices(fuse(dsk, ['c', 'd', 'e']))
+    result = optimize_slices(fuse(dsk, ['c', 'd', 'e'])[0])
 
     assert result == expected
 
diff --git a/dask/tests/test_optimize.py b/dask/tests/test_optimize.py
index fb713f7f3..749a03c83 100644
--- a/dask/tests/test_optimize.py
+++ b/dask/tests/test_optimize.py
@@ -18,27 +18,27 @@ def double(x):
 def test_cull():
     # 'out' depends on 'x' and 'y', but not 'z'
     d = {'x': 1, 'y': (inc, 'x'), 'z': (inc, 'x'), 'out': (add, 'y', 10)}
-    culled = cull(d, 'out')
+    culled, dependencies = cull(d, 'out')
     assert culled == {'x': 1, 'y': (inc, 'x'), 'out': (add, 'y', 10)}
+    assert dependencies == {'x': [], 'y': ['x'], 'out': ['y']}
+
     assert cull(d, 'out') == cull(d, ['out'])
-    assert cull(d, ['out', 'z']) == d
+    assert cull(d, ['out', 'z'])[0] == d
     assert cull(d, [['out'], ['z']]) == cull(d, ['out', 'z'])
     assert raises(KeyError, lambda: cull(d, 'badkey'))
 
 
 def test_fuse():
-    assert fuse({
-        'w': (inc, 'x'),
-        'x': (inc, 'y'),
-        'y': (inc, 'z'),
-        'z': (add, 'a', 'b'),
-        'a': 1,
-        'b': 2,
-    }) == {
-        'w': (inc, (inc, (inc, (add, 'a', 'b')))),
-        'a': 1,
-        'b': 2,
-    }
+    dsk, dependencies = fuse({'w': (inc, 'x'),
+                              'x': (inc, 'y'),
+                              'y': (inc, 'z'),
+                              'z': (add, 'a', 'b'),
+                              'a': 1,
+                              'b': 2})
+    assert dsk == {'w': (inc, (inc, (inc, (add, 'a', 'b')))),
+                   'a': 1,
+                   'b': 2}
+    assert dependencies == {'a': set(), 'b': set(), 'w': set(['a', 'b'])}
     assert fuse({
         'NEW': (inc, 'y'),
         'w': (inc, 'x'),
@@ -47,13 +47,16 @@ def test_fuse():
         'z': (add, 'a', 'b'),
         'a': 1,
         'b': 2,
-    }) == {
+    }) == ({
         'NEW': (inc, 'y'),
         'w': (inc, (inc, 'y')),
         'y': (inc, (add, 'a', 'b')),
         'a': 1,
         'b': 2,
-    }
+    },
+    {'a': set(), 'b': set(), 'y': set(['a', 'b']),
+     'w': set(['y']), 'NEW': set(['y'])})
+
     assert fuse({
         'v': (inc, 'y'),
         'u': (inc, 'w'),
@@ -65,13 +68,15 @@ def test_fuse():
         'b': (inc, 'd'),
         'c': 1,
         'd': 2,
-    }) == {
+    }) == ({
         'u': (inc, (inc, (inc, 'y'))),
         'v': (inc, 'y'),
         'y': (inc, (add, 'a', 'b')),
         'a': (inc, 1),
         'b': (inc, 2),
-    }
+    },
+    {'a': set(), 'b': set(), 'y': set(['a', 'b']),
+     'v': set(['y']), 'u': set(['y'])})
     assert fuse({
         'a': (inc, 'x'),
         'b': (inc, 'x'),
@@ -79,39 +84,43 @@ def test_fuse():
         'd': (inc, 'c'),
         'x': (inc, 'y'),
         'y': 0,
-    }) == {
+    }) == ({
         'a': (inc, 'x'),
         'b': (inc, 'x'),
         'd': (inc, (inc, 'x')),
         'x': (inc, 0),
-    }
+    },
+    {'x': set(), 'd': set(['x']), 'a': set(['x']), 'b': set(['x'])})
     assert fuse({
         'a': 1,
         'b': (inc, 'a'),
         'c': (add, 'b', 'b')
-    }) == {
+    }) == ({
         'b': (inc, 1),
         'c': (add, 'b', 'b')
-    }
+        }, {'b': set(), 'c': set(['b'])})
 
 
 def test_fuse_keys():
     assert (fuse({'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')}, keys=['b'])
-            == {'b': (inc, 1), 'c': (inc, 'b')})
-    assert fuse({
+            == ({'b': (inc, 1), 'c': (inc, 'b')},
+                {'b': set(), 'c': set(['b'])}))
+    dsk, dependencies = fuse({
         'w': (inc, 'x'),
         'x': (inc, 'y'),
         'y': (inc, 'z'),
         'z': (add, 'a', 'b'),
         'a': 1,
         'b': 2,
-    }, keys=['x', 'z']) == {
-        'w': (inc, 'x'),
-        'x': (inc, (inc, 'z')),
-        'z': (add, 'a', 'b'),
-        'a': 1,
-        'b': 2,
-    }
+    }, keys=['x', 'z'])
+
+    assert dsk == {'w': (inc, 'x'),
+                   'x': (inc, (inc, 'z')),
+                   'z': (add, 'a', 'b'),
+                   'a': 1,
+                   'b': 2 }
+    assert dependencies == {'a': set(), 'b': set(), 'z': set(['a', 'b']),
+                            'x': set(['z']), 'w': set(['x'])}
 
 
 def test_inline():
@@ -341,7 +350,7 @@ def test_fuse_getitem():
     dsk = {'x': (load, 'store', 'part', ['a', 'b']),
            'y': (getitem, 'x', 'a')}
     dsk2 = fuse_getitem(dsk, load, 3)
-    dsk2 = cull(dsk2, 'y')
+    dsk2, dependencies = cull(dsk2, 'y')
     assert dsk2 == {'y': (load, 'store', 'part', 'a')}
 
 
@@ -352,5 +361,5 @@ def test_fuse_selections():
            'y': (getitem, 'x', 'a')}
     merge = lambda t1, t2: (load, t2[1], t2[2], t1[2])
     dsk2 = fuse_selections(dsk, getitem, load, merge)
-    dsk2 = cull(dsk2, 'y')
+    dsk2, dependencies = cull(dsk2, 'y')
     assert dsk2 == {'y': (load, 'store', 'part', 'a')}
diff --git a/dask/tests/test_order.py b/dask/tests/test_order.py
index d64dee715..3b5f8c3ce 100644
--- a/dask/tests/test_order.py
+++ b/dask/tests/test_order.py
@@ -1,5 +1,6 @@
 from itertools import chain
-from dask.order import dfs, child_max, ndependents, order, inc, get_deps
+from dask.order import dfs, child_max, ndependents, order, inc
+from dask.core import get_deps
 
 
 def issorted(L, reverse=False):

pip install -e .[complete]
parse_log_pytest
numpy>=1.16.0 pandas>=1.0.0 cloudpickle partd distributed s3fs toolz psutil pytables bokeh bcolz scipy h5py ipython
pytest
apt-get update apt-get install -y graphviz liblzma-dev
3.5
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
aiobotocore @ file:///opt/conda/conda-bld/aiobotocore_1643638228694/work
aiohttp @ file:///tmp/build/80754af9/aiohttp_1632748060317/work
aioitertools @ file:///tmp/build/80754af9/aioitertools_1607109665762/work
async-timeout==3.0.1
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
backcall @ file:///home/ktietz/src/ci/backcall_1611930011877/work
bcolz==1.2.1
bokeh @ file:///tmp/build/80754af9/bokeh_1620710048147/work
boto3==1.23.10
botocore==1.26.10
brotlipy==0.7.0
certifi==2021.5.30
cffi @ file:///tmp/build/80754af9/cffi_1625814693874/work
chardet @ file:///tmp/build/80754af9/chardet_1607706739153/work
click==8.0.3
cloudpickle @ file:///tmp/build/80754af9/cloudpickle_1632508026186/work
contextvars==2.4
cryptography @ file:///tmp/build/80754af9/cryptography_1635366128178/work
cytoolz==0.11.0
-e git+https://github.com/dask/dask.git@b066b73cfa68e31848a0ff0d6348f6a0df73d4a7#egg=dask
decorator @ file:///opt/conda/conda-bld/decorator_1643638310831/work
distributed==1.9.5
fsspec @ file:///opt/conda/conda-bld/fsspec_1642510437511/work
h5py==2.10.0
HeapDict @ file:///Users/ktietz/demo/mc3/conda-bld/heapdict_1630598515714/work
idna @ file:///tmp/build/80754af9/idna_1637925883363/work
idna-ssl @ file:///tmp/build/80754af9/idna_ssl_1611752490495/work
immutables @ file:///tmp/build/80754af9/immutables_1628888996840/work
importlib-metadata==4.8.3
iniconfig==1.1.1
ipython @ file:///tmp/build/80754af9/ipython_1593447367857/work
ipython-genutils @ file:///tmp/build/80754af9/ipython_genutils_1606773439826/work
jedi @ file:///tmp/build/80754af9/jedi_1606932572482/work
Jinja2 @ file:///opt/conda/conda-bld/jinja2_1647436528585/work
jmespath @ file:///Users/ktietz/demo/mc3/conda-bld/jmespath_1630583964805/work
locket==0.2.1
MarkupSafe @ file:///tmp/build/80754af9/markupsafe_1621528150516/work
mock @ file:///tmp/build/80754af9/mock_1607622725907/work
msgpack @ file:///tmp/build/80754af9/msgpack-python_1612287171716/work
msgpack-python==0.5.6
multidict @ file:///tmp/build/80754af9/multidict_1607367768400/work
numexpr @ file:///tmp/build/80754af9/numexpr_1618853194344/work
numpy @ file:///tmp/build/80754af9/numpy_and_numpy_base_1603483703303/work
olefile @ file:///Users/ktietz/demo/mc3/conda-bld/olefile_1629805411829/work
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pandas==1.1.5
parso==0.7.0
partd @ file:///opt/conda/conda-bld/partd_1647245470509/work
pexpect @ file:///tmp/build/80754af9/pexpect_1605563209008/work
pickleshare @ file:///tmp/build/80754af9/pickleshare_1606932040724/work
Pillow @ file:///tmp/build/80754af9/pillow_1625670622947/work
pluggy==1.0.0
prompt-toolkit @ file:///tmp/build/80754af9/prompt-toolkit_1633440160888/work
psutil @ file:///tmp/build/80754af9/psutil_1612297621795/work
ptyprocess @ file:///tmp/build/80754af9/ptyprocess_1609355006118/work/dist/ptyprocess-0.7.0-py2.py3-none-any.whl
py==1.11.0
pycparser @ file:///tmp/build/80754af9/pycparser_1636541352034/work
Pygments @ file:///opt/conda/conda-bld/pygments_1644249106324/work
pyOpenSSL @ file:///opt/conda/conda-bld/pyopenssl_1643788558760/work
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
PySocks @ file:///tmp/build/80754af9/pysocks_1605305763431/work
pytest==7.0.1
python-dateutil @ file:///tmp/build/80754af9/python-dateutil_1626374649649/work
pytz==2021.3
PyYAML==5.4.1
s3fs==0.4.2
s3transfer==0.5.2
scipy @ file:///tmp/build/80754af9/scipy_1597686635649/work
six @ file:///tmp/build/80754af9/six_1644875935023/work
sortedcontainers @ file:///tmp/build/80754af9/sortedcontainers_1623949099177/work
tables==3.6.1
tblib @ file:///Users/ktietz/demo/mc3/conda-bld/tblib_1629402031467/work
tomli==1.2.3
toolz @ file:///tmp/build/80754af9/toolz_1636545406491/work
tornado @ file:///tmp/build/80754af9/tornado_1606942266872/work
traitlets @ file:///tmp/build/80754af9/traitlets_1632746497744/work
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
urllib3 @ file:///opt/conda/conda-bld/urllib3_1643638302206/work
wcwidth @ file:///Users/ktietz/demo/mc3/conda-bld/wcwidth_1629357192024/work
wrapt==1.12.1
yarl @ file:///tmp/build/80754af9/yarl_1606939915466/work
zict==2.0.0
zipp==3.6.0

name: dask
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - aiobotocore=2.1.0=pyhd3eb1b0_0
  - aiohttp=3.7.4.post0=py36h7f8727e_2
  - aioitertools=0.7.1=pyhd3eb1b0_0
  - async-timeout=3.0.1=py36h06a4308_0
  - attrs=21.4.0=pyhd3eb1b0_0
  - backcall=0.2.0=pyhd3eb1b0_0
  - bcolz=1.2.1=py36h04863e7_0
  - blas=1.0=openblas
  - blosc=1.21.3=h6a678d5_0
  - bokeh=2.3.2=py36h06a4308_0
  - brotlipy=0.7.0=py36h27cfd23_1003
  - bzip2=1.0.8=h5eee18b_6
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - cffi=1.14.6=py36h400218f_0
  - chardet=4.0.0=py36h06a4308_1003
  - click=8.0.3=pyhd3eb1b0_0
  - cloudpickle=2.0.0=pyhd3eb1b0_0
  - contextvars=2.4=py_0
  - cryptography=35.0.0=py36hd23ed53_0
  - cytoolz=0.11.0=py36h7b6447c_0
  - decorator=5.1.1=pyhd3eb1b0_0
  - freetype=2.12.1=h4a9f257_0
  - fsspec=2022.1.0=pyhd3eb1b0_0
  - giflib=5.2.2=h5eee18b_0
  - h5py=2.10.0=py36h7918eee_0
  - hdf5=1.10.4=hb1b8bf9_0
  - heapdict=1.0.1=pyhd3eb1b0_0
  - idna=3.3=pyhd3eb1b0_0
  - idna_ssl=1.1.0=py36h06a4308_0
  - immutables=0.16=py36h7f8727e_0
  - ipython=7.16.1=py36h5ca1d4c_0
  - ipython_genutils=0.2.0=pyhd3eb1b0_1
  - jedi=0.17.2=py36h06a4308_1
  - jinja2=3.0.3=pyhd3eb1b0_0
  - jmespath=0.10.0=pyhd3eb1b0_0
  - jpeg=9e=h5eee18b_3
  - lcms2=2.16=hb9589c4_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - lerc=4.0.0=h6a678d5_0
  - libdeflate=1.22=h5eee18b_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgfortran-ng=7.5.0=ha8ba4b0_17
  - libgfortran4=7.5.0=ha8ba4b0_17
  - libgomp=11.2.0=h1234567_1
  - libopenblas=0.3.18=hf726d26_0
  - libpng=1.6.39=h5eee18b_0
  - libstdcxx-ng=11.2.0=h1234567_1
  - libtiff=4.5.1=hffd6297_1
  - libwebp=1.2.4=h11a3e52_1
  - libwebp-base=1.2.4=h5eee18b_1
  - locket=0.2.1=py36h06a4308_1
  - lz4-c=1.9.4=h6a678d5_1
  - lzo=2.10=h7b6447c_2
  - markupsafe=2.0.1=py36h27cfd23_0
  - mock=4.0.3=pyhd3eb1b0_0
  - multidict=5.1.0=py36h27cfd23_2
  - ncurses=6.4=h6a678d5_0
  - numexpr=2.7.3=py36h4be448d_1
  - numpy=1.19.2=py36h6163131_0
  - numpy-base=1.19.2=py36h75fe3a5_0
  - olefile=0.46=pyhd3eb1b0_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pandas=1.1.5=py36ha9443f7_0
  - parso=0.7.0=py_0
  - partd=1.2.0=pyhd3eb1b0_1
  - pexpect=4.8.0=pyhd3eb1b0_3
  - pickleshare=0.7.5=pyhd3eb1b0_1003
  - pillow=8.3.1=py36h5aabda8_0
  - pip=21.2.2=py36h06a4308_0
  - prompt-toolkit=3.0.20=pyhd3eb1b0_0
  - psutil=5.8.0=py36h27cfd23_1
  - ptyprocess=0.7.0=pyhd3eb1b0_2
  - pycparser=2.21=pyhd3eb1b0_0
  - pygments=2.11.2=pyhd3eb1b0_0
  - pyopenssl=22.0.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pysocks=1.7.1=py36h06a4308_0
  - pytables=3.6.1=py36h71ec239_0
  - python=3.6.13=h12debd9_1
  - python-dateutil=2.8.2=pyhd3eb1b0_0
  - pytz=2021.3=pyhd3eb1b0_0
  - pyyaml=5.4.1=py36h27cfd23_1
  - readline=8.2=h5eee18b_0
  - scipy=1.5.2=py36habc2bb6_0
  - setuptools=58.0.4=py36h06a4308_0
  - six=1.16.0=pyhd3eb1b0_1
  - sortedcontainers=2.4.0=pyhd3eb1b0_0
  - sqlite=3.45.3=h5eee18b_0
  - tblib=1.7.0=pyhd3eb1b0_0
  - tk=8.6.14=h39e8969_0
  - toolz=0.11.2=pyhd3eb1b0_0
  - tornado=6.1=py36h27cfd23_0
  - traitlets=4.3.3=py36h06a4308_0
  - typing-extensions=4.1.1=hd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - urllib3=1.26.8=pyhd3eb1b0_0
  - wcwidth=0.2.5=pyhd3eb1b0_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - wrapt=1.12.1=py36h7b6447c_1
  - xz=5.6.4=h5eee18b_1
  - yaml=0.2.5=h7b6447c_0
  - yarl=1.6.3=py36h27cfd23_0
  - zict=2.0.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - zstd=1.5.6=hc292b87_0
  - pip:
      - boto3==1.23.10
      - botocore==1.26.10
      - distributed==1.9.5
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - msgpack-python==0.5.6
      - pluggy==1.0.0
      - py==1.11.0
      - pytest==7.0.1
      - s3fs==0.4.2
      - s3transfer==0.5.2
      - tomli==1.2.3
      - zipp==3.6.0
prefix: /opt/conda/envs/dask

dask/array/tests/test_optimization.py::test_optimize_slicing dask/tests/test_optimize.py::test_cull dask/tests/test_optimize.py::test_fuse dask/tests/test_optimize.py::test_fuse_keys dask/tests/test_optimize.py::test_fuse_getitem dask/tests/test_optimize.py::test_fuse_selections
dask/array/tests/test_optimization.py::test_fuse_getitem dask/array/tests/test_optimization.py::test_optimize_with_getitem_fusion dask/array/tests/test_optimization.py::test_fuse_slice dask/array/tests/test_optimization.py::test_fuse_slice_with_lists dask/array/tests/test_optimization.py::test_hard_fuse_slice_cases dask/array/tests/test_optimization.py::test_dont_fuse_different_slices dask/tests/test_optimize.py::test_inline dask/tests/test_optimize.py::test_inline_functions dask/tests/test_optimize.py::test_inline_ignores_curries_and_partials dask/tests/test_optimize.py::test_inline_doesnt_shrink_fast_functions_at_top dask/tests/test_optimize.py::test_inline_traverses_lists dask/tests/test_optimize.py::test_inline_protects_output_keys dask/tests/test_optimize.py::test_functions_of dask/tests/test_optimize.py::test_dealias dask/tests/test_optimize.py::test_dealias_keys dask/tests/test_optimize.py::test_equivalent dask/tests/test_optimize.py::test_equivalence_uncomparable dask/tests/test_optimize.py::test_sync_keys dask/tests/test_optimize.py::test_sync_uncomparable dask/tests/test_optimize.py::test_merge_sync dask/tests/test_order.py::test_ordering_keeps_groups_together dask/tests/test_order.py::test_prefer_broker_nodes dask/tests/test_order.py::test_base_of_reduce_preferred dask/tests/test_order.py::test_deep_bases_win_over_dependents dask/tests/test_order.py::test_prefer_deep dask/tests/test_order.py::test_stacklimit dask/tests/test_order.py::test_ndependents
dask/array/optimization.py dask/bag/core.py dask/dataframe/optimize.py dask/multiprocessing.py dask/order.py dask/dataframe/shuffle.py dask/optimize.py dask/core.py dask/async.py
dask/array/optimization.py dask/bag/core.py dask/dataframe/optimize.py dask/multiprocessing.py dask/order.py dask/dataframe/shuffle.py dask/optimize.py dask/core.py dask/async.py