diff --git a/.travis.yml b/.travis.yml
index 7baeae74e..9a5e9d5c5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -3,11 +3,11 @@ sudo: false
 
 env:
   matrix:
-    - PYTHON=2.7 NUMPY=1.10.4 PANDAS=0.19.0 COVERAGE='true' XTRATESTARGS=
-    - PYTHON=2.7 NUMPY=1.11.0 PANDAS=0.18.1 COVERAGE='false' PYTHONOPTIMIZE=2 XTRATESTARGS=--ignore=dask/diagnostics
-    - PYTHON=3.3 NUMPY=1.9.2 PANDAS=0.18.1 COVERAGE='false' XTRATESTARGS=
-    - PYTHON=3.4 NUMPY=1.10.4 PANDAS=0.18.0 COVERAGE='false' PYTHONOPTIMIZE=2 XTRATESTARGS=--ignore=dask/diagnostics
-    - PYTHON=3.5 NUMPY=1.11.0 PANDAS=0.19.0 COVERAGE='false' XTRATESTARGS=
+    - PYTHON=2.7 NUMPY=1.10.4 PANDAS=0.19.0 COVERAGE='true' PARALLEL='false' XTRATESTARGS=
+    - PYTHON=2.7 NUMPY=1.11.0 PANDAS=0.18.1 COVERAGE='false' PARALLEL='true' PYTHONOPTIMIZE=2 XTRATESTARGS=--ignore=dask/diagnostics
+    - PYTHON=3.3 NUMPY=1.9.2 PANDAS=0.18.1 COVERAGE='false' PARALLEL='true' XTRATESTARGS=
+    - PYTHON=3.4 NUMPY=1.10.4 PANDAS=0.18.0 COVERAGE='false' PARALLEL='true' PYTHONOPTIMIZE=2 XTRATESTARGS=--ignore=dask/diagnostics
+    - PYTHON=3.5 NUMPY=1.11.0 PANDAS=0.19.0 COVERAGE='false' PARALLEL='true' XTRATESTARGS=
 
 addons:
     apt:
@@ -35,11 +35,16 @@ install:
   - pip install blosc --upgrade
   - pip install graphviz moto flake8
   - if [[ $PYTHON < '3' ]]; then pip install git+https://github.com/Blosc/castra; fi
+  # For parallel testing (`-n` argument in XTRATESTARGS)
+  - pip install pytest-xdist
 
   # Install dask
   - pip install --no-deps -e .[complete]
 
 script:
+    # Need to make test order deterministic when parallelizing tests, hence PYTHONHASHSEED
+    # (see https://github.com/pytest-dev/pytest-xdist/issues/63)
+    - if [[ $PARALLEL == 'true' ]]; then export XTRATESTARGS="-n3 $XTRATESTARGS"; export PYTHONHASHSEED=42; fi
     - if [[ $COVERAGE == 'true' ]]; then coverage run $(which py.test) dask --runslow --doctest-modules --verbose $XTRATESTARGS; else py.test dask --runslow --verbose $XTRATESTARGS; fi
     - flake8 dask
 
diff --git a/dask/array/core.py b/dask/array/core.py
index e37806fb7..551be8537 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -17,7 +17,7 @@ import warnings
 
 from toolz.curried import (pipe, partition, concat, pluck, join, first,
                            memoize, map, groupby, valmap, accumulate, merge,
-                           reduce, interleave, sliding_window)
+                           reduce, interleave, sliding_window, assoc)
 import numpy as np
 
 from . import chunk
@@ -205,7 +205,6 @@ def broadcast_dimensions(argpairs, numblocks, sentinels=(1, (1,)),
 
     Parameters
     ----------
-
     argpairs: iterable
         name, ijk index pairs
     numblocks: dict
@@ -217,7 +216,6 @@ def broadcast_dimensions(argpairs, numblocks, sentinels=(1, (1,)),
 
     Examples
     --------
-
     >>> argpairs = [('x', 'ij'), ('y', 'ji')]
     >>> numblocks = {'x': (2, 3), 'y': (3, 2)}
     >>> broadcast_dimensions(argpairs, numblocks)
@@ -246,8 +244,8 @@ def broadcast_dimensions(argpairs, numblocks, sentinels=(1, (1,)),
 
     g2 = dict((k, v - set(sentinels) if len(v) > 1 else v) for k, v in g.items())
 
-    if consolidate is not None:
-        g2 = valmap(consolidate, g2)
+    if consolidate:
+        return valmap(consolidate, g2)
 
     if g2 and not set(map(len, g2.values())) == set([1]):
         raise ValueError("Shapes do not align %s" % g)
@@ -447,7 +445,7 @@ def map_blocks(func, *args, **kwargs):
     dtype: np.dtype
         Datatype of resulting array
     chunks: tuple (optional)
-        chunk shape of resulting blocks if the function does not preserve shape
+        Chunk shape of resulting blocks if the function does not preserve shape
     drop_axis: number or iterable (optional)
         Dimensions lost by the function
     new_axis: number or iterable (optional)
@@ -510,7 +508,7 @@ def map_blocks(func, *args, **kwargs):
     >>> y.numblocks
     (10,)
 
-    If these must match (up to broadcasting rules) then we can map arbitrary
+    If these match (up to broadcasting rules) then we can map arbitrary
     functions across blocks
 
     >>> def func(a, b):
@@ -551,63 +549,95 @@ def map_blocks(func, *args, **kwargs):
     if isinstance(new_axis, Number):
         new_axis = [new_axis]
 
+    if drop_axis and new_axis:
+        raise ValueError("Can't specify drop_axis and new_axis together")
+
     arrs = [a for a in args if isinstance(a, Array)]
     args = [(i, a) for i, a in enumerate(args) if not isinstance(a, Array)]
 
-    arginds = [(a, tuple(range(a.ndim))[::-1]) for a in arrs]
-
-    numblocks = dict([(a.name, a.numblocks) for a, _ in arginds])
-    argindsstr = list(concat([(a.name, ind) for a, ind in arginds]))
+    argpairs = [(a.name, tuple(range(a.ndim))[::-1]) for a in arrs]
+    numblocks = {a.name: a.numblocks for a in arrs}
+    arginds = list(concat(argpairs))
     out_ind = tuple(range(max(a.ndim for a in arrs)))[::-1]
 
+    try:
+        spec = getargspec(func)
+        block_id = ('block_id' in spec.args or
+                    'block_id' in getattr(spec, 'kwonly_args', ()))
+    except:
+        block_id = False
+
+    if block_id:
+        kwargs['block_id'] = '__dummy__'
+
     if args:
-        dsk = top(partial_by_order, name, out_ind, *argindsstr,
+        dsk = top(partial_by_order, name, out_ind, *arginds,
                   numblocks=numblocks, function=func, other=args,
                   **kwargs)
     else:
-        dsk = top(func, name, out_ind, *argindsstr, numblocks=numblocks,
+        dsk = top(func, name, out_ind, *arginds, numblocks=numblocks,
                   **kwargs)
 
-    # If func has block_id as an argument then swap out func
-    # for func with block_id partialed in
-    try:
-        spec = getargspec(func)
-    except:
-        spec = None
-    if spec:
-        args = spec.args
-        try:
-            args += spec.kwonlyargs
-        except AttributeError:
-            pass
-        if 'block_id' in args:
-            for k in dsk.keys():
-                dsk[k] = (partial(func, block_id=k[1:]),) + dsk[k][1:]
+    # If func has block_id as an argument, add it to the kwargs for each call
+    if block_id:
+        for k in dsk.keys():
+            dsk[k] = dsk[k][:-1] + (assoc(dsk[k][-1], 'block_id', k[1:]),)
 
-    numblocks = list(arrs[0].numblocks)
+    if len(arrs) == 1:
+        numblocks = list(arrs[0].numblocks)
+    else:
+        dims = broadcast_dimensions(argpairs, numblocks)
+        numblocks = [b for (_, b) in reversed(list(dims.items()))]
 
     if drop_axis:
+        if any(numblocks[i] > 1 for i in drop_axis):
+            raise ValueError("Can't drop an axis with more than 1 block. "
+                             "Please use `atop` instead.")
         dsk = dict((tuple(k for i, k in enumerate(k)
                           if i - 1 not in drop_axis), v)
                    for k, v in dsk.items())
         numblocks = [n for i, n in enumerate(numblocks) if i not in drop_axis]
-
-    if new_axis:
+    elif new_axis:
         dsk, old_dsk = dict(), dsk
         for key in old_dsk:
             new_key = list(key)
             for i in new_axis:
                 new_key.insert(i + 1, 0)
             dsk[tuple(new_key)] = old_dsk[key]
-        for i in sorted(new_axis, reverse=False):
+        for i in sorted(new_axis):
             numblocks.insert(i, 1)
 
-    if chunks is not None and chunks and not isinstance(chunks[0], tuple):
-        chunks = [nb * (bs,) for nb, bs in zip(numblocks, chunks)]
-    if chunks is not None:
-        chunks = tuple(chunks)
+    if chunks:
+        if len(chunks) != len(numblocks):
+            raise ValueError("Provided chunks have {0} dims, expected {1} "
+                             "dims.".format(len(chunks), len(numblocks)))
+        chunks2 = []
+        for i, (c, nb) in enumerate(zip(chunks, numblocks)):
+            if isinstance(c, tuple):
+                if not len(c) == nb:
+                    raise ValueError("Dimension {0} has {1} blocks, "
+                                     "chunks specified with "
+                                     "{2} blocks".format(i, nb, len(c)))
+                chunks2.append(c)
+            else:
+                chunks2.append(nb * (c,))
     else:
-        chunks = broadcast_chunks(*[a.chunks for a in arrs])
+        if len(arrs) == 1:
+            chunks2 = list(arrs[0].chunks)
+        else:
+            try:
+                chunks2 = list(broadcast_chunks(*[a.chunks for a in arrs]))
+            except:
+                raise ValueError("Arrays in `map_blocks` don't align, can't "
+                                 "infer output chunks. Please provide "
+                                 "`chunks` kwarg.")
+        if drop_axis:
+            chunks2 = [c for (i, c) in enumerate(chunks2) if i not in drop_axis]
+        elif new_axis:
+            for i in sorted(new_axis):
+                chunks2.insert(i, (1,))
+
+    chunks = tuple(chunks2)
 
     return Array(merge(dsk, *[a.dask for a in arrs]), name, chunks, dtype)
 
@@ -1726,19 +1756,49 @@ def common_blockdim(blockdims):
     --------
 
     >>> common_blockdim([(3,), (2, 1)])
-    set([(2, 1)])
+    (2, 1)
+    >>> common_blockdim([(1, 2), (2, 1)])
+    (1, 1, 1)
     >>> common_blockdim([(2, 2), (3, 1)])  # doctest: +SKIP
     Traceback (most recent call last):
         ...
     ValueError: Chunks do not align
     """
     non_trivial_dims = set([d for d in blockdims if len(d) > 1])
-    if len(non_trivial_dims) > 1:
-        raise ValueError('Chunks do not align %s' % non_trivial_dims)
-    elif non_trivial_dims:
-        return non_trivial_dims
-    else:
-        return blockdims
+    if len(non_trivial_dims) == 1:
+        return first(non_trivial_dims)
+    if len(non_trivial_dims) == 0:
+        return max(blockdims, key=first)
+
+    if len(set(map(sum, non_trivial_dims))) > 1:
+        raise ValueError("Chunks do not add up to same value", blockdims)
+
+    # We have multiple non-trivial chunks on this axis
+    # e.g. (5, 2) and (4, 3)
+
+    # We create a single chunk tuple with the same total length
+    # that evenly divides both, e.g. (4, 1, 2)
+
+    # To accomplish this we walk down all chunk tuples together, finding the
+    # smallest element, adding it to the output, and subtracting it from all
+    # other elements and remove the element itself.  We stop once we have
+    # burned through all of the chunk tuples.
+    # For efficiency's sake we reverse the lists so that we can pop off the end
+    rchunks = [list(ntd)[::-1] for ntd in non_trivial_dims]
+    total = sum(first(non_trivial_dims))
+    i = 0
+
+    out = []
+    while i < total:
+        m = min(c[-1] for c in rchunks)
+        out.append(m)
+        for c in rchunks:
+            c[-1] -= m
+            if c[-1] == 0:
+                c.pop()
+        i += m
+
+    return tuple(out)
 
 
 def unify_chunks(*args):
@@ -1766,6 +1826,12 @@ def unify_chunks(*args):
 
     chunkss = broadcast_dimensions(nameinds, blockdim_dict,
                                    consolidate=common_blockdim)
+    max_parts = max(arg.npartitions for arg in args[::2])
+    nparts = np.prod(list(map(len, chunkss.values())))
+
+    if nparts >= max_parts * 10:
+        warnings.warn("Increasing number of chunks by factor of %d" %
+                      (nparts / max_parts))
     arrays = [a.rechunk(tuple(chunkss[j] if a.shape[n] > 1 else 1
                               for n, j in enumerate(i)))
               for a, i in arginds]
diff --git a/dask/async.py b/dask/async.py
index 930133f8f..a26611eff 100644
--- a/dask/async.py
+++ b/dask/async.py
@@ -117,6 +117,9 @@ from __future__ import absolute_import, division, print_function
 import sys
 import traceback
 
+from toolz import identity
+
+from .compatibility import Queue
 from .core import (istask, flatten, reverse_dict, get_dependencies, ishashable,
                    has_tasks)
 from .context import _globals
@@ -251,7 +254,7 @@ def _execute_task(arg, cache, dsk=None):
         return arg
 
 
-def execute_task(key, task, data, queue, get_id, raise_on_exception=False):
+def execute_task(key, task_info, dumps, loads, get_id, raise_on_exception=False):
     """
     Compute task and handle all administration
 
@@ -260,23 +263,24 @@ def execute_task(key, task, data, queue, get_id, raise_on_exception=False):
     _execute_task - actually execute task
     """
     try:
+        task, data = loads(task_info)
         result = _execute_task(task, data)
         id = get_id()
-        result = key, result, None, id
-    except Exception as e:
-        if raise_on_exception:
-            raise
-        exc_type, exc_value, exc_traceback = sys.exc_info()
-        tb = ''.join(traceback.format_tb(exc_traceback))
-        result = key, e, tb, None
-    try:
-        queue.put(result)
+        result = dumps((result, None, id))
     except Exception as e:
         if raise_on_exception:
             raise
         exc_type, exc_value, exc_traceback = sys.exc_info()
         tb = ''.join(traceback.format_tb(exc_traceback))
-        queue.put((key, e, tb, None))
+        try:
+            result = dumps((e, tb, None))
+        except Exception as e:
+            if raise_on_exception:
+                raise
+            exc_type, exc_value, exc_traceback = sys.exc_info()
+            tb = ''.join(traceback.format_tb(exc_traceback))
+            result = dumps((e, tb, None))
+    return key, result
 
 
 def release_data(key, state, delete=True):
@@ -372,8 +376,10 @@ The main function of the scheduler.  Get is the main entry point.
 
 
 def get_async(apply_async, num_workers, dsk, result, cache=None,
-              queue=None, get_id=default_get_id, raise_on_exception=False,
-              rerun_exceptions_locally=None, callbacks=None, **kwargs):
+              get_id=default_get_id, raise_on_exception=False,
+              rerun_exceptions_locally=None, callbacks=None,
+              dumps=identity, loads=identity,
+              **kwargs):
     """ Asynchronous get function
 
     This is a general version of various asynchronous schedulers for dask.  It
@@ -400,6 +406,11 @@ def get_async(apply_async, num_workers, dsk, result, cache=None,
     rerun_exceptions_locally : bool, optional
         Whether to rerun failing tasks in local process to enable debugging
         (False by default)
+    dumps: callable, optional
+        Function to serialize task data and results to communicate between
+        worker and parent.  Defaults to identity.
+    loads: callable, optional
+        Inverse function of `dumps`.  Defaults to identity.
     callbacks : tuple or list of tuples, optional
         Callbacks are passed in as tuples of length 5. Multiple sets of
         callbacks may be passed in as a list of tuples. For more information,
@@ -410,7 +421,7 @@ def get_async(apply_async, num_workers, dsk, result, cache=None,
 
     threaded.get
     """
-    assert queue
+    queue = Queue()
 
     if callbacks is None:
         callbacks = _globals['callbacks']
@@ -458,8 +469,10 @@ def get_async(apply_async, num_workers, dsk, result, cache=None,
             data = dict((dep, state['cache'][dep])
                         for dep in get_dependencies(dsk, key))
             # Submit
-            apply_async(execute_task, args=[key, dsk[key], data, queue,
-                                            get_id, raise_on_exception])
+            apply_async(execute_task,
+                        args=(key, dumps((dsk[key], data)),
+                              dumps, loads, get_id, raise_on_exception),
+                        callback=queue.put)
 
         # Seed initial tasks into the thread pool
         while state['ready'] and len(state['running']) < num_workers:
@@ -467,7 +480,14 @@ def get_async(apply_async, num_workers, dsk, result, cache=None,
 
         # Main loop, wait on tasks to finish, insert new ones
         while state['waiting'] or state['ready'] or state['running']:
-            key, res, tb, worker_id = queue.get()
+            key, res_info = queue.get()
+            try:
+                res, tb, worker_id = loads(res_info)
+            except Exception:
+                for _, _, _, _, finish in callbacks:
+                    if finish:
+                        finish(dsk, state, True)
+                raise
             if isinstance(res, Exception):
                 for _, _, _, _, finish in callbacks:
                     if finish:
@@ -483,8 +503,10 @@ def get_async(apply_async, num_workers, dsk, result, cache=None,
             finish_task(dsk, key, state, results, keyorder.get)
             for f in posttask_cbs:
                 f(key, res, dsk, state, worker_id)
+
             while state['ready'] and len(state['running']) < num_workers:
                 fire_task()
+
     except KeyboardInterrupt:
         for cb in started_cbs:
             if cb[-1]:
@@ -510,9 +532,11 @@ GIL
 """
 
 
-def apply_sync(func, args=(), kwds={}):
+def apply_sync(func, args=(), kwds={}, callback=None):
     """ A naive synchronous version of apply_async """
-    return func(*args, **kwds)
+    res = func(*args, **kwds)
+    if callback is not None:
+        callback(res)
 
 
 def get_sync(dsk, keys, **kwargs):
@@ -520,10 +544,8 @@ def get_sync(dsk, keys, **kwargs):
 
     Can be useful for debugging.
     """
-    from .compatibility import Queue
     kwargs.pop('num_workers', None)    # if num_workers present, remove it
-    queue = Queue()
-    return get_async(apply_sync, 1, dsk, keys, queue=queue,
+    return get_async(apply_sync, 1, dsk, keys,
                      raise_on_exception=True, **kwargs)
 
 
diff --git a/dask/multiprocessing.py b/dask/multiprocessing.py
index ac7cea83b..ff9bf66d0 100644
--- a/dask/multiprocessing.py
+++ b/dask/multiprocessing.py
@@ -9,7 +9,6 @@ from .context import _globals
 from .optimize import fuse, cull
 
 import cloudpickle
-from toolz import curry
 
 
 if sys.version_info.major < 3:
@@ -63,13 +62,6 @@ def get(dsk, keys, num_workers=None, func_loads=None, func_dumps=None,
     else:
         cleanup = False
 
-    manager = multiprocessing.Manager()
-    queue = manager.Queue()
-
-    apply_async = pickle_apply_async(pool.apply_async,
-                                     func_dumps=func_dumps,
-                                     func_loads=func_loads)
-
     # Optimize Dask
     dsk2, dependencies = cull(dsk, keys)
     if optimize_graph:
@@ -77,50 +69,20 @@ def get(dsk, keys, num_workers=None, func_loads=None, func_dumps=None,
     else:
         dsk3 = dsk2
 
+    # We specify marshalling functions in order to catch serialization
+    # errors and report them to the user.
+    loads = func_loads or _globals.get('func_loads') or _loads
+    dumps = func_dumps or _globals.get('func_dumps') or _dumps
+
+    # Note former versions used a multiprocessing Manager to share
+    # a Queue between parent and workers, but this is fragile on Windows
+    # (issue #1652).
     try:
         # Run
-        result = get_async(apply_async, len(pool._pool), dsk3, keys,
-                           queue=queue, get_id=_process_get_id, **kwargs)
+        result = get_async(pool.apply_async, len(pool._pool), dsk3, keys,
+                           get_id=_process_get_id,
+                           dumps=dumps, loads=loads, **kwargs)
     finally:
         if cleanup:
             pool.close()
     return result
-
-
-def apply_func(sfunc, may_fail, wont_fail, loads=None):
-    loads = loads or _globals.get('loads') or _loads
-    func = loads(sfunc)
-    key, queue, get_id, raise_on_exception = loads(wont_fail)
-    try:
-        task, data = loads(may_fail)
-    except Exception as e:
-        # Need a new reference for the exception, as `e` falls out of scope in
-        # python 3
-        exception = e
-
-        def serialization_failure():
-            raise exception
-
-        task = (serialization_failure,)
-        data = {}
-
-    return func(key, task, data, queue, get_id,
-                raise_on_exception=raise_on_exception)
-
-
-@curry
-def pickle_apply_async(apply_async, func, args=(),
-                       func_loads=None, func_dumps=None):
-    # XXX: To deal with deserialization errors of tasks, this version of
-    # apply_async doesn't actually match that of `pool.apply_async`. It's
-    # customized to fit the signature of `dask.async.execute_task`, which is
-    # the only function ever actually passed as `func`. This is a bit of a
-    # hack, but it works pretty well. If the signature of `execute_task`
-    # changes, then this will need to be changed as well.
-    dumps = func_dumps or _globals.get('func_dumps') or _dumps
-    key, task, data, queue, get_id, raise_on_exception = args
-    sfunc = dumps(func)
-    may_fail = dumps((task, data))
-    wont_fail = dumps((key, queue, get_id, raise_on_exception))
-    return apply_async(curry(apply_func, loads=func_loads),
-                       args=[sfunc, may_fail, wont_fail])
diff --git a/dask/threaded.py b/dask/threaded.py
index d1b96fc05..c2fffc1d8 100644
--- a/dask/threaded.py
+++ b/dask/threaded.py
@@ -11,7 +11,6 @@ import threading
 from threading import current_thread, Lock
 
 from .async import get_async
-from .compatibility import Queue
 from .context import _globals
 from .utils_test import inc, add  # noqa: F401
 
@@ -65,9 +64,8 @@ def get(dsk, result, cache=None, num_workers=None, **kwargs):
                 pool = ThreadPool(num_workers)
                 pools[thread][num_workers] = pool
 
-    queue = Queue()
     results = get_async(pool.apply_async, len(pool._pool), dsk, result,
-                        cache=cache, queue=queue, get_id=_thread_get_id,
+                        cache=cache, get_id=_thread_get_id,
                         **kwargs)
 
     # Cleanup pools associated to dead threads
diff --git a/dask/utils.py b/dask/utils.py
index 9c14d3a7c..5f7f38257 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -118,6 +118,28 @@ def filetext(text, extension='', open=open, mode='w'):
         yield filename
 
 
+@contextmanager
+def changed_cwd(new_cwd):
+    old_cwd = os.getcwd()
+    os.chdir(new_cwd)
+    try:
+        yield
+    finally:
+        os.chdir(old_cwd)
+
+
+@contextmanager
+def tmp_cwd(dir=None):
+    with tmpdir(dir) as dirname:
+        with changed_cwd(dirname):
+            yield dirname
+
+
+@contextmanager
+def noop_context():
+    yield
+
+
 def repr_long_list(seq):
     """
 
@@ -150,27 +172,32 @@ class IndexCallable(object):
 
 
 @contextmanager
-def filetexts(d, open=open, mode='t'):
+def filetexts(d, open=open, mode='t', use_tmpdir=True):
     """ Dumps a number of textfiles to disk
 
     d - dict
         a mapping from filename to text like {'a.csv': '1,1\n2,2'}
+
+    Since this is meant for use in tests, this context manager will
+    automatically switch to a temporary current directory, to avoid
+    race conditions when running tests in parallel.
     """
-    for filename, text in d.items():
-        f = open(filename, 'w' + mode)
-        try:
-            f.write(text)
-        finally:
+    with (tmp_cwd() if use_tmpdir else noop_context()):
+        for filename, text in d.items():
+            f = open(filename, 'w' + mode)
             try:
-                f.close()
-            except AttributeError:
-                pass
-
-    yield list(d)
-
-    for filename in d:
-        if os.path.exists(filename):
-            os.remove(filename)
+                f.write(text)
+            finally:
+                try:
+                    f.close()
+                except AttributeError:
+                    pass
+
+        yield list(d)
+
+        for filename in d:
+            if os.path.exists(filename):
+                os.remove(filename)
 
 
 compressions = {'gz': 'gzip', 'bz2': 'bz2', 'xz': 'xz'}

Test failures when parallel testing
This is on Linux. To reproduce: install the xdist plugin with pip (see https://pytest.org/dev/xdist.html#installation-of-xdist-plugin) then run something like `PYTHONHASHSEED=1 py.test -n=auto --tb=short dask`.

These are the failures I get here:
```
_______________________________________________________________ test_read_text[xz-None-ascii1] ________________________________________________________________
[gw0] linux -- Python 3.5.2 /home/antoine/miniconda3/envs/dask35/bin/python
dask/bag/tests/test_text.py:41: in test_read_text
    L, = compute(b)
dask/base.py:171: in compute
    results = get(dsk, keys, **kwargs)
dask/async.py:549: in get_sync
    raise_on_exception=True, **kwargs)
dask/async.py:478: in get_async
    fire_task()
dask/async.py:474: in fire_task
    callback=queue.put)
dask/async.py:535: in apply_sync
    res = func(*args, **kwds)
dask/async.py:266: in execute_task
    result = _execute_task(task, data)
dask/async.py:247: in _execute_task
    return func(*args2)
../miniconda3/envs/dask35/lib/python3.5/lzma.py:210: in read1
    return self._buffer.read1(size)
../miniconda3/envs/dask35/lib/python3.5/_compression.py:68: in readinto
    data = self.read(len(byte_view))
../miniconda3/envs/dask35/lib/python3.5/_compression.py:103: in read
    data = self._decompressor.decompress(rawblock, size)
E   _lzma.LZMAError: Input format not supported by decoder
_______________________________________________________________________ test_read_bytes _______________________________________________________________________
[gw0] linux -- Python 3.5.2 /home/antoine/miniconda3/envs/dask35/bin/python
dask/bytes/tests/test_local.py:40: in test_read_bytes
    results = compute(*concat(values))
dask/base.py:171: in compute
    results = get(dsk, keys, **kwargs)
dask/async.py:549: in get_sync
    raise_on_exception=True, **kwargs)
dask/async.py:478: in get_async
    fire_task()
dask/async.py:474: in fire_task
    callback=queue.put)
dask/async.py:535: in apply_sync
    res = func(*args, **kwds)
dask/async.py:266: in execute_task
    result = _execute_task(task, data)
dask/async.py:247: in _execute_task
    return func(*args2)
dask/bytes/local.py:74: in read_block_from_file
    with open(path, 'rb') as f:
E   FileNotFoundError: [Errno 2] No such file or directory: '/home/antoine/dask/.test.accounts.2.json'
```

diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index 887abac6e..c852b3cd2 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -8,6 +8,7 @@ import time
 from distutils.version import LooseVersion
 from operator import add, sub, getitem
 from threading import Lock
+import warnings
 
 from toolz import merge, countby, concat
 from toolz.curried import identity
@@ -29,7 +30,8 @@ from dask.array.core import (getem, getarray, getarray_nofancy, top, dotmany,
                              broadcast_to, reshape, fromfunction,
                              blockdims_from_blockshape, store, optimize,
                              from_func, normalize_chunks, broadcast_chunks,
-                             atop, from_delayed, concatenate_axes)
+                             atop, from_delayed, concatenate_axes,
+                             common_blockdim)
 from dask.array.utils import assert_eq
 
 # temporary until numpy functions migrated
@@ -780,8 +782,8 @@ def test_map_blocks2():
     x = np.arange(10, dtype='i8')
     d = from_array(x, chunks=(2,))
 
-    def func(block, block_id=None):
-        return np.ones_like(block) * sum(block_id)
+    def func(block, block_id=None, c=0):
+        return np.ones_like(block) * sum(block_id) + c
 
     out = d.map_blocks(func, dtype='i8')
     expected = np.array([0, 0, 1, 1, 2, 2, 3, 3, 4, 4], dtype='i8')
@@ -789,6 +791,12 @@ def test_map_blocks2():
     assert_eq(out, expected)
     assert same_keys(d.map_blocks(func, dtype='i8'), out)
 
+    out = d.map_blocks(func, dtype='i8', c=1)
+    expected = expected + 1
+
+    assert_eq(out, expected)
+    assert same_keys(d.map_blocks(func, dtype='i8', c=1), out)
+
 
 def test_map_blocks_with_constants():
     d = da.arange(10, chunks=3)
@@ -1852,15 +1860,38 @@ def test_map_blocks_with_changed_dimension():
 
     e = d.map_blocks(lambda b: b.sum(axis=0), chunks=(4,), drop_axis=0,
                      dtype=d.dtype)
-    assert e.ndim == 1
     assert e.chunks == ((4, 4),)
     assert_eq(e, x.sum(axis=0))
 
+    # Provided chunks have wrong shape
+    with pytest.raises(ValueError):
+        d.map_blocks(lambda b: b.sum(axis=0), chunks=(7, 4), drop_axis=0)
+
+    with pytest.raises(ValueError):
+        d.map_blocks(lambda b: b.sum(axis=0), chunks=((4, 4, 4),), drop_axis=0)
+
+    # Can't drop axis with more than 1 block
+    with pytest.raises(ValueError):
+        d.map_blocks(lambda b: b.sum(axis=1), drop_axis=1, dtype=d.dtype)
+
+    # Can't use both drop_axis and new_axis
+    with pytest.raises(ValueError):
+        d.map_blocks(lambda b: b, drop_axis=1, new_axis=1)
+
+    d = da.from_array(x, chunks=(4, 8))
+    e = d.map_blocks(lambda b: b.sum(axis=1), drop_axis=1, dtype=d.dtype)
+    assert e.chunks == ((4, 3),)
+    assert_eq(e, x.sum(axis=1))
+
     x = np.arange(64).reshape((8, 8))
     d = da.from_array(x, chunks=(4, 4))
     e = d.map_blocks(lambda b: b[None, :, :, None],
                      chunks=(1, 4, 4, 1), new_axis=[0, 3], dtype=d.dtype)
-    assert e.ndim == 4
+    assert e.chunks == ((1,), (4, 4), (4, 4), (1,))
+    assert_eq(e, x[None, :, :, None])
+
+    e = d.map_blocks(lambda b: b[None, :, :, None],
+                     new_axis=[0, 3], dtype=d.dtype)
     assert e.chunks == ((1,), (4, 4), (4, 4), (1,))
     assert_eq(e, x[None, :, :, None])
 
@@ -2235,3 +2266,61 @@ def test_atop_concatenate():
     z = atop(f, 'j', x, 'ijk', y, 'ki', y, 'ij', concatenate=True,
              dtype=x._dtype)
     assert_eq(z, np.ones(10))
+
+
+def test_common_blockdim():
+    assert common_blockdim([(5,), (5,)]) == (5,)
+    assert common_blockdim([(5,), (2, 3,)]) == (2, 3)
+    assert common_blockdim([(5, 5), (2, 3, 5)]) == (2, 3, 5)
+    assert common_blockdim([(5, 5), (2, 3, 5)]) == (2, 3, 5)
+    assert common_blockdim([(5, 2, 3), (2, 3, 5)]) == (2, 3, 2, 3)
+
+    assert common_blockdim([(1, 2), (2, 1)]) == (1, 1, 1)
+    assert common_blockdim([(1, 2, 2), (2, 1, 2), (2, 2, 1)]) == (1, 1, 1, 1, 1)
+
+
+def test_uneven_chunks_that_fit_neatly():
+    x = da.arange(10, chunks=((5, 5),))
+    y = da.ones(10, chunks=((5, 2, 3),))
+
+    assert_eq(x + y, np.arange(10) + np.ones(10))
+
+    z = x + y
+    assert z.chunks == ((5, 2, 3),)
+
+
+def test_elemwise_uneven_chunks():
+    x = da.arange(10, chunks=((4, 6),))
+    y = da.ones(10, chunks=((6, 4),))
+
+    assert_eq(x + y, np.arange(10) + np.ones(10))
+
+    z = x + y
+    assert z.chunks == ((4, 2, 4),)
+
+    x = da.random.random((10, 10), chunks=((4, 6), (5, 2, 3)))
+    y = da.random.random((4, 10, 10), chunks=((2, 2), (6, 4), (2, 3, 5)))
+
+    z = x + y
+    assert_eq(x + y, x.compute() + y.compute())
+    assert z.chunks == ((2, 2), (4, 2, 4), (2, 3, 2, 3))
+
+
+def test_uneven_chunks_atop():
+    x = da.random.random((10, 10), chunks=((2, 3, 2, 3), (5, 5)))
+    y = da.random.random((10, 10), chunks=((4, 4, 2), (4, 2, 4)))
+    z = atop(np.dot, 'ik', x, 'ij', y, 'jk', dtype=x._dtype, concatenate=True)
+    assert z.chunks == (x.chunks[0], y.chunks[1])
+
+    assert_eq(z, x.compute().dot(y))
+
+
+def test_warn_bad_rechunking():
+    x = da.ones((20, 20), chunks=(20, 1))
+    y = da.ones((20, 20), chunks=(1, 20))
+
+    with warnings.catch_warnings(record=True) as record:
+        x + y
+
+    assert record
+    assert '20' in record[0].message.args[0]

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
botocore @ file:///opt/conda/conda-bld/botocore_1642672735464/work
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@f14ec6a4ec0d4fb9ff2499ac53d0df4116465c8a#egg=dask
decorator @ file:///opt/conda/conda-bld/decorator_1643638310831/work
distributed==1.13.3
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 @ file:///opt/conda/conda-bld/s3fs_1643701468749/work
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
  - botocore=1.23.24=pyhd3eb1b0_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
  - s3fs=2022.1.0=pyhd3eb1b0_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:
      - distributed==1.13.3
      - 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
      - tomli==1.2.3
      - zipp==3.6.0
prefix: /opt/conda/envs/dask

dask/array/tests/test_array_core.py::test_map_blocks2 dask/array/tests/test_array_core.py::test_map_blocks_with_changed_dimension dask/array/tests/test_array_core.py::test_common_blockdim dask/array/tests/test_array_core.py::test_uneven_chunks_that_fit_neatly dask/array/tests/test_array_core.py::test_elemwise_uneven_chunks dask/array/tests/test_array_core.py::test_uneven_chunks_atop dask/array/tests/test_array_core.py::test_warn_bad_rechunking
dask/array/tests/test_array_core.py::test_field_access dask/array/tests/test_array_core.py::test_field_access_with_shape dask/array/tests/test_array_core.py::test_coarsen dask/array/tests/test_array_core.py::test_coarsen_with_excess
dask/array/tests/test_array_core.py::test_getem dask/array/tests/test_array_core.py::test_top dask/array/tests/test_array_core.py::test_top_supports_broadcasting_rules dask/array/tests/test_array_core.py::test_concatenate3_on_scalars dask/array/tests/test_array_core.py::test_chunked_dot_product dask/array/tests/test_array_core.py::test_chunked_transpose_plus_one dask/array/tests/test_array_core.py::test_transpose dask/array/tests/test_array_core.py::test_broadcast_dimensions_works_with_singleton_dimensions dask/array/tests/test_array_core.py::test_broadcast_dimensions dask/array/tests/test_array_core.py::test_Array dask/array/tests/test_array_core.py::test_uneven_chunks dask/array/tests/test_array_core.py::test_numblocks_suppoorts_singleton_block_dims dask/array/tests/test_array_core.py::test_keys dask/array/tests/test_array_core.py::test_Array_computation dask/array/tests/test_array_core.py::test_stack dask/array/tests/test_array_core.py::test_short_stack dask/array/tests/test_array_core.py::test_stack_scalars dask/array/tests/test_array_core.py::test_concatenate dask/array/tests/test_array_core.py::test_concatenate_fixlen_strings dask/array/tests/test_array_core.py::test_vstack dask/array/tests/test_array_core.py::test_hstack dask/array/tests/test_array_core.py::test_dstack dask/array/tests/test_array_core.py::test_take dask/array/tests/test_array_core.py::test_compress dask/array/tests/test_array_core.py::test_binops dask/array/tests/test_array_core.py::test_isnull dask/array/tests/test_array_core.py::test_isclose dask/array/tests/test_array_core.py::test_broadcast_shapes dask/array/tests/test_array_core.py::test_elemwise_on_scalars dask/array/tests/test_array_core.py::test_partial_by_order dask/array/tests/test_array_core.py::test_elemwise_with_ndarrays dask/array/tests/test_array_core.py::test_elemwise_differently_chunked dask/array/tests/test_array_core.py::test_operators dask/array/tests/test_array_core.py::test_operator_dtype_promotion dask/array/tests/test_array_core.py::test_tensordot dask/array/tests/test_array_core.py::test_dot_method dask/array/tests/test_array_core.py::test_T dask/array/tests/test_array_core.py::test_norm dask/array/tests/test_array_core.py::test_choose dask/array/tests/test_array_core.py::test_where dask/array/tests/test_array_core.py::test_where_has_informative_error dask/array/tests/test_array_core.py::test_insert dask/array/tests/test_array_core.py::test_multi_insert dask/array/tests/test_array_core.py::test_broadcast_to dask/array/tests/test_array_core.py::test_ravel dask/array/tests/test_array_core.py::test_reshape dask/array/tests/test_array_core.py::test_reshape_fails_for_dask_only dask/array/tests/test_array_core.py::test_reshape_unknown_dimensions dask/array/tests/test_array_core.py::test_full dask/array/tests/test_array_core.py::test_map_blocks dask/array/tests/test_array_core.py::test_map_blocks_with_constants dask/array/tests/test_array_core.py::test_map_blocks_with_kwargs dask/array/tests/test_array_core.py::test_fromfunction dask/array/tests/test_array_core.py::test_from_function_requires_block_args dask/array/tests/test_array_core.py::test_repr dask/array/tests/test_array_core.py::test_slicing_with_ellipsis dask/array/tests/test_array_core.py::test_slicing_with_ndarray dask/array/tests/test_array_core.py::test_dtype dask/array/tests/test_array_core.py::test_blockdims_from_blockshape dask/array/tests/test_array_core.py::test_coerce dask/array/tests/test_array_core.py::test_store dask/array/tests/test_array_core.py::test_store_compute_false dask/array/tests/test_array_core.py::test_store_locks dask/array/tests/test_array_core.py::test_to_hdf5 dask/array/tests/test_array_core.py::test_np_array_with_zero_dimensions dask/array/tests/test_array_core.py::test_unique dask/array/tests/test_array_core.py::test_dtype_complex dask/array/tests/test_array_core.py::test_astype dask/array/tests/test_array_core.py::test_arithmetic dask/array/tests/test_array_core.py::test_clip dask/array/tests/test_array_core.py::test_elemwise_consistent_names dask/array/tests/test_array_core.py::test_optimize dask/array/tests/test_array_core.py::test_slicing_with_non_ndarrays dask/array/tests/test_array_core.py::test_getarray dask/array/tests/test_array_core.py::test_squeeze dask/array/tests/test_array_core.py::test_size dask/array/tests/test_array_core.py::test_nbytes dask/array/tests/test_array_core.py::test_Array_normalizes_dtype dask/array/tests/test_array_core.py::test_args dask/array/tests/test_array_core.py::test_from_array_with_lock dask/array/tests/test_array_core.py::test_from_array_slicing_results_in_ndarray dask/array/tests/test_array_core.py::test_from_func dask/array/tests/test_array_core.py::test_topk dask/array/tests/test_array_core.py::test_topk_k_bigger_than_chunk dask/array/tests/test_array_core.py::test_bincount dask/array/tests/test_array_core.py::test_bincount_with_weights dask/array/tests/test_array_core.py::test_bincount_raises_informative_error_on_missing_minlength_kwarg dask/array/tests/test_array_core.py::test_digitize dask/array/tests/test_array_core.py::test_histogram dask/array/tests/test_array_core.py::test_histogram_alternative_bins_range dask/array/tests/test_array_core.py::test_histogram_return_type dask/array/tests/test_array_core.py::test_histogram_extra_args_and_shapes dask/array/tests/test_array_core.py::test_concatenate3_2 dask/array/tests/test_array_core.py::test_map_blocks3 dask/array/tests/test_array_core.py::test_from_array_with_missing_chunks dask/array/tests/test_array_core.py::test_cache dask/array/tests/test_array_core.py::test_take_dask_from_numpy dask/array/tests/test_array_core.py::test_normalize_chunks dask/array/tests/test_array_core.py::test_raise_on_no_chunks dask/array/tests/test_array_core.py::test_chunks_is_immutable dask/array/tests/test_array_core.py::test_raise_on_bad_kwargs dask/array/tests/test_array_core.py::test_long_slice dask/array/tests/test_array_core.py::test_h5py_newaxis dask/array/tests/test_array_core.py::test_ellipsis_slicing dask/array/tests/test_array_core.py::test_point_slicing dask/array/tests/test_array_core.py::test_point_slicing_with_full_slice dask/array/tests/test_array_core.py::test_slice_with_floats dask/array/tests/test_array_core.py::test_vindex_errors dask/array/tests/test_array_core.py::test_vindex_merge dask/array/tests/test_array_core.py::test_empty_array dask/array/tests/test_array_core.py::test_array dask/array/tests/test_array_core.py::test_cov dask/array/tests/test_array_core.py::test_corrcoef dask/array/tests/test_array_core.py::test_memmap dask/array/tests/test_array_core.py::test_to_npy_stack dask/array/tests/test_array_core.py::test_view dask/array/tests/test_array_core.py::test_view_fortran dask/array/tests/test_array_core.py::test_h5py_tokenize dask/array/tests/test_array_core.py::test_broadcast_chunks dask/array/tests/test_array_core.py::test_chunks_error dask/array/tests/test_array_core.py::test_array_compute_forward_kwargs dask/array/tests/test_array_core.py::test_dont_fuse_outputs dask/array/tests/test_array_core.py::test_dont_dealias_outputs dask/array/tests/test_array_core.py::test_timedelta_op dask/array/tests/test_array_core.py::test_to_delayed dask/array/tests/test_array_core.py::test_cumulative dask/array/tests/test_array_core.py::test_eye dask/array/tests/test_array_core.py::test_diag dask/array/tests/test_array_core.py::test_tril_triu dask/array/tests/test_array_core.py::test_tril_triu_errors dask/array/tests/test_array_core.py::test_atop_names dask/array/tests/test_array_core.py::test_atop_new_axes dask/array/tests/test_array_core.py::test_atop_kwargs dask/array/tests/test_array_core.py::test_from_delayed dask/array/tests/test_array_core.py::test_A_property dask/array/tests/test_array_core.py::test_copy dask/array/tests/test_array_core.py::test_npartitions dask/array/tests/test_array_core.py::test_astype_gh1151 dask/array/tests/test_array_core.py::test_elemwise_name dask/array/tests/test_array_core.py::test_map_blocks_name dask/array/tests/test_array_core.py::test_from_array_names dask/array/tests/test_array_core.py::test_array_picklable dask/array/tests/test_array_core.py::test_swapaxes dask/array/tests/test_array_core.py::test_from_array_raises_on_bad_chunks dask/array/tests/test_array_core.py::test_concatenate_axes dask/array/tests/test_array_core.py::test_atop_concatenate
dask/multiprocessing.py .travis.yml dask/threaded.py dask/array/core.py dask/async.py dask/utils.py
dask/multiprocessing.py .travis.yml dask/threaded.py dask/array/core.py dask/async.py dask/utils.py