Repository: dask/dask

Issue:
Fusing purely linear chains not working
On branch `master`, in particular after https://github.com/dask/dask/pull/3979 was merged: I am trying to make use of the new feature of merging linear chains in (sub)graphs. To do so, I enable it via the config:

In `config.yaml`:
```
fuse_subgraphs: True
```

The following works as expected, with the new `fuse_subgraphs` option the long chains are now fused:

```python
import dask
import dask.array as da
import numpy as np

if __name__ == '__main__':
    x = da.from_array(np.zeros((3,)), chunks=(1,))
    x = x + x
    branch = x + x
    x = x + branch
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    dask.visualize(x, filename='fused.svg', optimize_graph=True)
```

However, graphs without any branches will not fuse:

```python
import dask
import dask.array as da
import numpy as np

if __name__ == '__main__':
    x = da.from_array(np.zeros((3,)), chunks=(1,))
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    x = x + x
    dask.visualize(x, filename='not-fused.svg', optimize_graph=True)
```

I have tracked the issue back to https://github.com/dask/dask/blob/master/dask/optimization.py#L570 --- for a linear graph this returns early, skipping fusion of linear chains. I do not understand what exactly is done in this part of the code, so I cannot propose a precise fix.

Discussion:
jcrist: Planning on merging tomorrow if no comment.
mrocklin: I don't really know what's going on here, but tests pass so I'm happy.  +1 to merge

Patch:
diff --git a/dask/optimization.py b/dask/optimization.py
index e2a181f76..c3af897ee 100644
--- a/dask/optimization.py
+++ b/dask/optimization.py
@@ -567,7 +567,10 @@ def fuse(dsk, keys=None, dependencies=None, ave_width=None, max_width=None,
     reducible = {k for k, vals in rdeps.items() if len(vals) == 1}
     if keys:
         reducible -= keys
-    if not reducible:
+    if (not reducible and
+            (not fuse_subgraphs or all(len(set(v)) != 1 for v in rdeps.values()))):
+        # Quick return if there's nothing to do. Only progress if there's tasks
+        # fusible by the main `fuse`, or by `fuse_subgraphs` if enabled.
         return dsk, deps
 
     rv = dsk.copy()


Test patch:
diff --git a/dask/tests/test_optimization.py b/dask/tests/test_optimization.py
index 304ab8374..df7164357 100644
--- a/dask/tests/test_optimization.py
+++ b/dask/tests/test_optimization.py
@@ -1200,3 +1200,27 @@ def test_fuse_subgraphs():
             'inc-6': (inc, (inc, 'add-2'))
         }))
     assert res in sols
+
+
+def test_fuse_subgraphs_linear_chains_of_duplicate_deps():
+    dsk = {'x-1': 1,
+           'add-1': (add, 'x-1', 'x-1'),
+           'add-2': (add, 'add-1', 'add-1'),
+           'add-3': (add, 'add-2', 'add-2'),
+           'add-4': (add, 'add-3', 'add-3'),
+           'add-5': (add, 'add-4', 'add-4')}
+
+    res = fuse(dsk, 'add-5', fuse_subgraphs=True)
+    sol = with_deps({
+        'add-x-1': (
+            SubgraphCallable({
+                'x-1': 1,
+                'add-1': (add, 'x-1', 'x-1'),
+                'add-2': (add, 'add-1', 'add-1'),
+                'add-3': (add, 'add-2', 'add-2'),
+                'add-4': (add, 'add-3', 'add-3'),
+                'add-5': (add, 'add-4', 'add-4')
+            }, 'add-5', ()),),
+        'add-5': 'add-x-1'
+    })
+    assert res == sol