Repository: dask/dask

Issue:
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'
```


Discussion:

Patch:
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 patch:
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]