We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
dask test failure
it seems `atop` is using an older dask API

```
================================================================================= FAILURES ==================================================================================
____________________________________________________________________________ test_compute[expr0] ____________________________________________________________________________
blaze/compute/tests/test_dask.py:69: in test_compute
    result = compute(expr, dask_ns)
../../../../miniconda/envs/py27/lib/python2.7/site-packages/multipledispatch/dispatcher.py:163: in __call__
    return func(*args, **kwargs)
blaze/compute/core.py:470: in compute
    result = top_then_bottom_then_top_again_etc(expr3, d4, **kwargs)
blaze/compute/core.py:164: in top_then_bottom_then_top_again_etc
    expr2, scope2 = bottom_up_until_type_break(expr, scope, **kwargs)
blaze/compute/core.py:371: in bottom_up_until_type_break
    **kwargs)}
../../../../miniconda/envs/py27/lib/python2.7/site-packages/multipledispatch/dispatcher.py:163: in __call__
    return func(*args, **kwargs)
blaze/compute/dask.py:40: in compute_broadcast
    *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))
../../../../code/py/dask/dask/array/core.py:1099: in atop
    numblocks = dict([(a.name, a.numblocks) for a, ind in arginds])
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `blaze/compute/dask.py`
Content:
```
1 from __future__ import absolute_import, division, print_function
2 
3 from numbers import Number
4 from toolz import concat, first, curry, compose
5 from datashape import DataShape
6 
7 from blaze import compute, ndim
8 from blaze.dispatch import dispatch
9 from blaze.compute.core import compute_up, optimize
10 from blaze.expr import (ElemWise, symbol, Reduction, Transpose, TensorDot,
11                         Expr, Slice, Broadcast)
12 from blaze.expr.split import split
13 
14 from dask.array.core import (_concatenate2, Array, atop, names, transpose,
15                              tensordot)
16 
17 
18 def compute_it(expr, leaves, *data, **kwargs):
19     kwargs.pop('scope')
20     return compute(expr, dict(zip(leaves, data)), **kwargs)
21 
22 
23 def elemwise_array(expr, *data, **kwargs):
24     leaves = expr._inputs
25     expr_inds = tuple(range(ndim(expr)))[::-1]
26     return atop(curry(compute_it, expr, leaves, **kwargs),
27                 next(names), expr_inds,
28                 *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))
29 
30 
31 try:
32     from blaze.compute.numba import (get_numba_ufunc, broadcast_collect,
33             Broadcastable)
34 
35     def compute_broadcast(expr, *data, **kwargs):
36         expr_inds = tuple(range(ndim(expr)))[::-1]
37         func = get_numba_ufunc(expr)
38         return atop(func,
39                     next(names), expr_inds,
40                     *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))
41 
42     def optimize_array(expr, *data):
43         return broadcast_collect(expr, Broadcastable=Broadcastable,
44                                        WantToBroadcast=Broadcastable)
45 
46     for i in range(5):
47         compute_up.register(Broadcast, *([(Array, Number)] * i))(compute_broadcast)
48         optimize.register(Expr, *([(Array, Number)] * i))(optimize_array)
49 
50 except ImportError:
51     pass
52 
53 
54 for i in range(5):
55     compute_up.register(ElemWise, *([Array] * i))(elemwise_array)
56 
57 
58 @dispatch(Reduction, Array)
59 def compute_up(expr, data, **kwargs):
60     leaf = expr._leaves()[0]
61     chunk = symbol('chunk', DataShape(*(tuple(map(first, data.chunks)) +
62                                         (leaf.dshape.measure,))))
63     (chunk, chunk_expr), (agg, agg_expr) = split(expr._child, expr,
64                                                  chunk=chunk)
65 
66     inds = tuple(range(ndim(leaf)))
67     tmp = atop(curry(compute_it, chunk_expr, [chunk], **kwargs),
68                next(names), inds,
69                data, inds)
70 
71     return atop(compose(curry(compute_it, agg_expr, [agg], **kwargs),
72                         curry(_concatenate2, axes=expr.axis)),
73                 next(names), tuple(i for i in inds if i not in expr.axis),
74                 tmp, inds)
75 
76 
77 @dispatch(Transpose, Array)
78 def compute_up(expr, data, **kwargs):
79     return transpose(data, expr.axes)
80 
81 
82 @dispatch(TensorDot, Array, Array)
83 def compute_up(expr, lhs, rhs, **kwargs):
84     return tensordot(lhs, rhs, (expr._left_axes, expr._right_axes))
85 
86 
87 @dispatch(Slice, Array)
88 def compute_up(expr, data, **kwargs):
89     return data[expr.index]
90
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/blaze/compute/dask.py b/blaze/compute/dask.py
--- a/blaze/compute/dask.py
+++ b/blaze/compute/dask.py
@@ -24,7 +24,7 @@
     leaves = expr._inputs
     expr_inds = tuple(range(ndim(expr)))[::-1]
     return atop(curry(compute_it, expr, leaves, **kwargs),
-                next(names), expr_inds,
+                expr_inds,
                 *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))
 
 
@@ -36,7 +36,7 @@
         expr_inds = tuple(range(ndim(expr)))[::-1]
         func = get_numba_ufunc(expr)
         return atop(func,
-                    next(names), expr_inds,
+                    expr_inds,
                     *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))
 
     def optimize_array(expr, *data):
@@ -64,13 +64,12 @@
                                                  chunk=chunk)
 
     inds = tuple(range(ndim(leaf)))
-    tmp = atop(curry(compute_it, chunk_expr, [chunk], **kwargs),
-               next(names), inds,
-               data, inds)
+    tmp = atop(curry(compute_it, chunk_expr, [chunk], **kwargs), inds, data,
+               inds)
 
     return atop(compose(curry(compute_it, agg_expr, [agg], **kwargs),
                         curry(_concatenate2, axes=expr.axis)),
-                next(names), tuple(i for i in inds if i not in expr.axis),
+                tuple(i for i in inds if i not in expr.axis),
                 tmp, inds)

{"golden_diff": "diff --git a/blaze/compute/dask.py b/blaze/compute/dask.py\n--- a/blaze/compute/dask.py\n+++ b/blaze/compute/dask.py\n@@ -24,7 +24,7 @@\n     leaves = expr._inputs\n     expr_inds = tuple(range(ndim(expr)))[::-1]\n     return atop(curry(compute_it, expr, leaves, **kwargs),\n-                next(names), expr_inds,\n+                expr_inds,\n                 *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))\n \n \n@@ -36,7 +36,7 @@\n         expr_inds = tuple(range(ndim(expr)))[::-1]\n         func = get_numba_ufunc(expr)\n         return atop(func,\n-                    next(names), expr_inds,\n+                    expr_inds,\n                     *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))\n \n     def optimize_array(expr, *data):\n@@ -64,13 +64,12 @@\n                                                  chunk=chunk)\n \n     inds = tuple(range(ndim(leaf)))\n-    tmp = atop(curry(compute_it, chunk_expr, [chunk], **kwargs),\n-               next(names), inds,\n-               data, inds)\n+    tmp = atop(curry(compute_it, chunk_expr, [chunk], **kwargs), inds, data,\n+               inds)\n \n     return atop(compose(curry(compute_it, agg_expr, [agg], **kwargs),\n                         curry(_concatenate2, axes=expr.axis)),\n-                next(names), tuple(i for i in inds if i not in expr.axis),\n+                tuple(i for i in inds if i not in expr.axis),\n                 tmp, inds)\n", "issue": "dask test failure\nit seems `atop` is using an older dask API\n\n```\n================================================================================= FAILURES ==================================================================================\n____________________________________________________________________________ test_compute[expr0] ____________________________________________________________________________\nblaze/compute/tests/test_dask.py:69: in test_compute\n    result = compute(expr, dask_ns)\n../../../../miniconda/envs/py27/lib/python2.7/site-packages/multipledispatch/dispatcher.py:163: in __call__\n    return func(*args, **kwargs)\nblaze/compute/core.py:470: in compute\n    result = top_then_bottom_then_top_again_etc(expr3, d4, **kwargs)\nblaze/compute/core.py:164: in top_then_bottom_then_top_again_etc\n    expr2, scope2 = bottom_up_until_type_break(expr, scope, **kwargs)\nblaze/compute/core.py:371: in bottom_up_until_type_break\n    **kwargs)}\n../../../../miniconda/envs/py27/lib/python2.7/site-packages/multipledispatch/dispatcher.py:163: in __call__\n    return func(*args, **kwargs)\nblaze/compute/dask.py:40: in compute_broadcast\n    *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))\n../../../../code/py/dask/dask/array/core.py:1099: in atop\n    numblocks = dict([(a.name, a.numblocks) for a, ind in arginds])\n```\n\n", "before_files": [{"content": "from __future__ import absolute_import, division, print_function\n\nfrom numbers import Number\nfrom toolz import concat, first, curry, compose\nfrom datashape import DataShape\n\nfrom blaze import compute, ndim\nfrom blaze.dispatch import dispatch\nfrom blaze.compute.core import compute_up, optimize\nfrom blaze.expr import (ElemWise, symbol, Reduction, Transpose, TensorDot,\n                        Expr, Slice, Broadcast)\nfrom blaze.expr.split import split\n\nfrom dask.array.core import (_concatenate2, Array, atop, names, transpose,\n                             tensordot)\n\n\ndef compute_it(expr, leaves, *data, **kwargs):\n    kwargs.pop('scope')\n    return compute(expr, dict(zip(leaves, data)), **kwargs)\n\n\ndef elemwise_array(expr, *data, **kwargs):\n    leaves = expr._inputs\n    expr_inds = tuple(range(ndim(expr)))[::-1]\n    return atop(curry(compute_it, expr, leaves, **kwargs),\n                next(names), expr_inds,\n                *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))\n\n\ntry:\n    from blaze.compute.numba import (get_numba_ufunc, broadcast_collect,\n            Broadcastable)\n\n    def compute_broadcast(expr, *data, **kwargs):\n        expr_inds = tuple(range(ndim(expr)))[::-1]\n        func = get_numba_ufunc(expr)\n        return atop(func,\n                    next(names), expr_inds,\n                    *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))\n\n    def optimize_array(expr, *data):\n        return broadcast_collect(expr, Broadcastable=Broadcastable,\n                                       WantToBroadcast=Broadcastable)\n\n    for i in range(5):\n        compute_up.register(Broadcast, *([(Array, Number)] * i))(compute_broadcast)\n        optimize.register(Expr, *([(Array, Number)] * i))(optimize_array)\n\nexcept ImportError:\n    pass\n\n\nfor i in range(5):\n    compute_up.register(ElemWise, *([Array] * i))(elemwise_array)\n\n\n@dispatch(Reduction, Array)\ndef compute_up(expr, data, **kwargs):\n    leaf = expr._leaves()[0]\n    chunk = symbol('chunk', DataShape(*(tuple(map(first, data.chunks)) +\n                                        (leaf.dshape.measure,))))\n    (chunk, chunk_expr), (agg, agg_expr) = split(expr._child, expr,\n                                                 chunk=chunk)\n\n    inds = tuple(range(ndim(leaf)))\n    tmp = atop(curry(compute_it, chunk_expr, [chunk], **kwargs),\n               next(names), inds,\n               data, inds)\n\n    return atop(compose(curry(compute_it, agg_expr, [agg], **kwargs),\n                        curry(_concatenate2, axes=expr.axis)),\n                next(names), tuple(i for i in inds if i not in expr.axis),\n                tmp, inds)\n\n\n@dispatch(Transpose, Array)\ndef compute_up(expr, data, **kwargs):\n    return transpose(data, expr.axes)\n\n\n@dispatch(TensorDot, Array, Array)\ndef compute_up(expr, lhs, rhs, **kwargs):\n    return tensordot(lhs, rhs, (expr._left_axes, expr._right_axes))\n\n\n@dispatch(Slice, Array)\ndef compute_up(expr, data, **kwargs):\n    return data[expr.index]\n", "path": "blaze/compute/dask.py"}], "after_files": [{"content": "from __future__ import absolute_import, division, print_function\n\nfrom numbers import Number\nfrom toolz import concat, first, curry, compose\nfrom datashape import DataShape\n\nfrom blaze import compute, ndim\nfrom blaze.dispatch import dispatch\nfrom blaze.compute.core import compute_up, optimize\nfrom blaze.expr import (ElemWise, symbol, Reduction, Transpose, TensorDot,\n                        Expr, Slice, Broadcast)\nfrom blaze.expr.split import split\n\nfrom dask.array.core import (_concatenate2, Array, atop, names, transpose,\n                             tensordot)\n\n\ndef compute_it(expr, leaves, *data, **kwargs):\n    kwargs.pop('scope')\n    return compute(expr, dict(zip(leaves, data)), **kwargs)\n\n\ndef elemwise_array(expr, *data, **kwargs):\n    leaves = expr._inputs\n    expr_inds = tuple(range(ndim(expr)))[::-1]\n    return atop(curry(compute_it, expr, leaves, **kwargs),\n                expr_inds,\n                *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))\n\n\ntry:\n    from blaze.compute.numba import (get_numba_ufunc, broadcast_collect,\n            Broadcastable)\n\n    def compute_broadcast(expr, *data, **kwargs):\n        expr_inds = tuple(range(ndim(expr)))[::-1]\n        func = get_numba_ufunc(expr)\n        return atop(func,\n                    expr_inds,\n                    *concat((dat, tuple(range(ndim(dat))[::-1])) for dat in data))\n\n    def optimize_array(expr, *data):\n        return broadcast_collect(expr, Broadcastable=Broadcastable,\n                                       WantToBroadcast=Broadcastable)\n\n    for i in range(5):\n        compute_up.register(Broadcast, *([(Array, Number)] * i))(compute_broadcast)\n        optimize.register(Expr, *([(Array, Number)] * i))(optimize_array)\n\nexcept ImportError:\n    pass\n\n\nfor i in range(5):\n    compute_up.register(ElemWise, *([Array] * i))(elemwise_array)\n\n\n@dispatch(Reduction, Array)\ndef compute_up(expr, data, **kwargs):\n    leaf = expr._leaves()[0]\n    chunk = symbol('chunk', DataShape(*(tuple(map(first, data.chunks)) +\n                                        (leaf.dshape.measure,))))\n    (chunk, chunk_expr), (agg, agg_expr) = split(expr._child, expr,\n                                                 chunk=chunk)\n\n    inds = tuple(range(ndim(leaf)))\n    tmp = atop(curry(compute_it, chunk_expr, [chunk], **kwargs), inds, data,\n               inds)\n\n    return atop(compose(curry(compute_it, agg_expr, [agg], **kwargs),\n                        curry(_concatenate2, axes=expr.axis)),\n                tuple(i for i in inds if i not in expr.axis),\n                tmp, inds)\n\n\n@dispatch(Transpose, Array)\ndef compute_up(expr, data, **kwargs):\n    return transpose(data, expr.axes)\n\n\n@dispatch(TensorDot, Array, Array)\ndef compute_up(expr, lhs, rhs, **kwargs):\n    return tensordot(lhs, rhs, (expr._left_axes, expr._right_axes))\n\n\n@dispatch(Slice, Array)\ndef compute_up(expr, data, **kwargs):\n    return data[expr.index]\n", "path": "blaze/compute/dask.py"}]}