Repository: dask/dask

Issue:
Bug: threaded scheduler is not interruptible
```
import dask, time
dask.__version__
> u'0.14.1'

# can be interrupted
fun = dask.delayed(lambda x: time.sleep(5))
dask.compute([fun(x) for x in range(100)], get = dask.threaded.get)

# can't be interrupted
fun2 = dask.delayed(lambda x: sum([1 for x in range(10000000)]))
dask.compute([fun2(x) for x in range(100)], get = dask.threaded.get)
```

Using the threaded scheduler, I noticed that some tasks cannot be interrupted by pressing Ctrl+C in a python console. Directly after pressing Ctrl+C the CPU usage of that process jumps up: It uses all available minus one cores. Judging by the output of top, the process seems to spend the computing time in the kernel space. 



Discussion:
jcrist: I couldn't think of a good way to test this, but also don't really think it needs a test. We'd want to raise a `KeyboardInterrupt` in the main thread while the scheduler (also running in the main thread) is blocked by a call to `Queue.get`. Not sure if this is possible.

Patch:
diff --git a/.travis.yml b/.travis.yml
index 4e2415057..d878274f6 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,9 +5,11 @@ env:
   matrix:
     - PYTHON=2.7 NUMPY=1.10.4 PANDAS=0.19.0 COVERAGE='true' PARALLEL='false' TEST_IMPORTS='false' XTRATESTARGS=
     - PYTHON=2.7 NUMPY=1.11.0 PANDAS=0.19.2 COVERAGE='false' PARALLEL='true' TEST_IMPORTS='false' PYTHONOPTIMIZE=2 XTRATESTARGS=--ignore=dask/diagnostics
+    - PYTHON=2.7 NUMPY=1.12.1 PANDAS=0.19.2 COVERAGE='false' PARALLEL='true' TEST_IMPORTS='false' PYTHONOPTIMIZE=2 XTRATESTARGS=--ignore=dask/diagnostics
     - PYTHON=3.4 NUMPY=1.10.4 PANDAS=0.19.1 COVERAGE='false' PARALLEL='true' TEST_IMPORTS='false' PYTHONOPTIMIZE=2 XTRATESTARGS=--ignore=dask/diagnostics
     - PYTHON=3.5 NUMPY=1.11.0 PANDAS=0.19.2 COVERAGE='false' PARALLEL='true' TEST_IMPORTS='false' XTRATESTARGS=
     - PYTHON=3.6 NUMPY=1.11.2 PANDAS=0.19.2 COVERAGE='false' PARALLEL='true' TEST_IMPORTS='true' XTRATESTARGS=
+    - PYTHON=3.6 NUMPY=1.12.1 PANDAS=0.19.2 COVERAGE='false' PARALLEL='true' TEST_IMPORTS='true' XTRATESTARGS=
 
 addons:
     apt:
diff --git a/appveyor.yml b/appveyor.yml
index b0837c310..0f54df9e2 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -12,7 +12,7 @@ environment:
     # Since appveyor is quite slow, we only use a single configuration
     - PYTHON: "3.5"
       ARCH: "64"
-      NUMPY: "1.11"
+      NUMPY: "1.12"
       PANDAS: "0.19.2"
       CONDA_ENV: testenv
 
diff --git a/dask/array/__init__.py b/dask/array/__init__.py
index 23cd3800a..57a6fd8f0 100644
--- a/dask/array/__init__.py
+++ b/dask/array/__init__.py
@@ -3,7 +3,7 @@ from __future__ import absolute_import, division, print_function
 from ..utils import ignoring
 from .core import (Array, stack, concatenate, take, tensordot, transpose,
         from_array, choose, where, coarsen, insert, broadcast_to, ravel,
-        roll, fromfunction, unique, store, squeeze, topk, bincount,
+        roll, fromfunction, unique, store, squeeze, topk, bincount, tile,
         digitize, histogram, map_blocks, atop, to_hdf5, dot, cov, array,
         dstack, vstack, hstack, to_npy_stack, from_npy_stack, compress,
         from_delayed, round, swapaxes, repeat, asarray)
diff --git a/dask/array/core.py b/dask/array/core.py
index 365d832a8..b0d9d229f 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2618,6 +2618,8 @@ def elemwise(op, *args, **kwargs):
         msg = "%s does not take the following keyword arguments %s"
         raise TypeError(msg % (op.__name__, str(sorted(set(kwargs) - set(['name', 'dtype'])))))
 
+    args = [np.asarray(a) if isinstance(a, (list, tuple)) else a for a in args]
+
     shapes = [getattr(arg, 'shape', ()) for arg in args]
     shapes = [s if isinstance(s, Iterable) else () for s in shapes]
     out_ndim = len(broadcast_shapes(*shapes))   # Raises ValueError if dimensions mismatch
@@ -2839,7 +2841,7 @@ def roll(array, shift, axis=None):
         result = ravel(result)
 
         if not isinstance(shift, Integral):
-            TypeError(
+            raise TypeError(
                 "Expect `shift` to be an instance of Integral"
                 " when `axis` is None."
             )
@@ -2856,6 +2858,9 @@ def roll(array, shift, axis=None):
         except TypeError:
             axis = (axis,)
 
+    if len(shift) != len(axis):
+        raise ValueError("Must have the same number of shifts as axes.")
+
     for i, s in zip(axis, shift):
         s = -s
         s %= result.shape[i]
@@ -3794,7 +3799,7 @@ def swapaxes(a, axis1, axis2):
                 dtype=a.dtype)
 
 
-@wraps(np.dot)
+@wraps(np.repeat)
 def repeat(a, repeats, axis=None):
     if axis is None:
         if a.ndim == 1:
@@ -3835,6 +3840,21 @@ def repeat(a, repeats, axis=None):
     return concatenate(out, axis=axis)
 
 
+@wraps(np.tile)
+def tile(A, reps):
+    if not isinstance(reps, Integral):
+        raise NotImplementedError("Only integer valued `reps` supported.")
+
+    if reps < 0:
+        raise ValueError("Negative `reps` are not allowed.")
+    elif reps == 0:
+        return A[..., :0]
+    elif reps == 1:
+        return A
+
+    return concatenate(reps * [A], axis=-1)
+
+
 def slice_with_dask_array(x, index):
     y = elemwise(getitem, x, index, dtype=x.dtype)
 
diff --git a/dask/async.py b/dask/async.py
index e933b889f..058f2e73f 100644
--- a/dask/async.py
+++ b/dask/async.py
@@ -114,10 +114,11 @@ See the function ``inline_functions`` for more information.
 """
 from __future__ import absolute_import, division, print_function
 
+import os
 import sys
 import traceback
 
-from .compatibility import Queue
+from .compatibility import Queue, Empty
 from .core import (istask, flatten, reverse_dict, get_dependencies, ishashable,
                    has_tasks)
 from .context import _globals
@@ -127,6 +128,30 @@ from .optimize import cull
 from .utils_test import add, inc  # noqa: F401
 
 
+if sys.version_info.major < 3:
+    # Due to a bug in python 2.7 Queue.get, if a timeout isn't specified then
+    # `Queue.get` can't be interrupted. A workaround is to specify an extremely
+    # long timeout, which then allows it to be interrupted.
+    # For more information see: https://bugs.python.org/issue1360
+    def queue_get(q):
+        return q.get(block=True, timeout=(365 * 24 * 60 * 60))
+
+elif os.name == 'nt':
+    # Python 3 windows Queue.get also doesn't handle interrupts properly. To
+    # workaround this we poll at a sufficiently large interval that it
+    # shouldn't affect performance, but small enough that users trying to kill
+    # an application shouldn't care.
+    def queue_get(q):
+        while True:
+            try:
+                return q.get(block=True, timeout=0.1)
+            except Empty:
+                pass
+else:
+    def queue_get(q):
+        return q.get()
+
+
 DEBUG = False
 
 
@@ -488,7 +513,7 @@ 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_info = queue.get()
+            key, res_info = queue_get(queue)
             res, tb, worker_id = loads(res_info)
             if isinstance(res, BaseException):
                 if rerun_exceptions_locally:
diff --git a/docs/source/array-api.rst b/docs/source/array-api.rst
index 4750f8b0d..32503530b 100644
--- a/docs/source/array-api.rst
+++ b/docs/source/array-api.rst
@@ -127,6 +127,7 @@ Top level user functions:
    tan
    tanh
    tensordot
+   tile
    topk
    transpose
    tril
diff --git a/docs/source/debugging.rst b/docs/source/debugging.rst
new file mode 100644
index 000000000..105a7857e
--- /dev/null
+++ b/docs/source/debugging.rst
@@ -0,0 +1,238 @@
+Debugging
+=========
+
+Debugging parallel programs is hard.  Normal debugging tools like logging and
+using ``pdb`` to interact with tracebacks stop working normally when exceptions
+occur in far-away machines or different processes or threads.
+
+Dask has a variety of mechanisms to make this process easier.  Depending on
+your situation some of these approaches may be more appropriate than others.
+
+These approaches are ordered from lightweight or easy solutions to more
+involved solutions.
+
+Exceptions
+----------
+
+When a task in your computation fails the standard way of understanding what
+went wrong is to look at the exception and traceback.  Often people do this
+with the ``pdb`` module, IPython ``%debug`` or ``%pdb`` magics, or by just
+looking at the traceback and investigating where in their code the exception
+occurred.
+
+Normally when a computation computes in a separate thread or a different
+machine these approaches break down.  Dask provides a few mechanisms to
+recreate the normal Python debugging experience.
+
+Inspect Exceptions and Tracebacks
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+By default, Dask already copies the exception and traceback wherever they
+occur and reraises that exception locally.  If your task failed with a
+``ZeroDivisionError`` remotely then you'll get a ``ZeroDivisionError`` in your
+interactive session.  Similarly you'll see a full traceback of where this error
+occurred, which, just like in normal Python, can help you to identify the
+troublsome spot in your code.
+
+However, you cannot use the ``pdb`` module or ``%debug`` IPython magics with
+these tracebacks to look at the value of variables during failure.  You can
+only inspect things visually.  Additionally, the top of the traceback may be
+filled with functions that are dask-specific and not relevant to your
+problem, you can safely ignore these.
+
+Both the single-machine and distributed schedulers do this.
+
+
+Use the Single-Threaded Scheduler
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Dask ships with a simple single-threaded scheduler.  This doesn't offer any
+parallel performance improvements, but does run your Dask computation
+faithfully in your local thread, allowing you to use normal tools like ``pdb``,
+``%debug`` IPython magics, the profiling tools like the ``cProfile`` module and
+`snakeviz <https://jiffyclub.github.io/snakeviz/>`_.  This allows you to use
+all of your normal Python debugging tricks in Dask computations, as long as you
+don't need parallelism.
+
+This only works for single-machine schedulers.  It does not work with
+dask.distributed unless you are comfortable using the Tornado API (look at the
+`testing infrastructure
+<http://distributed.readthedocs.io/en/latest/develop.html#writing-tests>`_
+docs, which accomplish this).  Also, because this operates on a single machine
+it assumes that your computation can run on a single machine without exceeding
+memory limits.  It may be wise to use this approach on smaller versions of your
+problem if possible.
+
+
+Rerun Failed Task Locally
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+If a remote task fails, we can collect the function and all inputs, bring them
+to the local thread, and then rerun the function in hopes of triggering the
+same exception locally, where normal debugging tools can be used.
+
+With the single machine schedulers, use the ``rerun_exceptions_locally=True``
+keyword.
+
+.. code-block:: python
+
+   x.compute(rerun_exceptions_locally=True)
+
+On the distributed scheduler use the ``recreate_error_locally`` method on
+anything that contains ``Futures`` :
+
+.. code-block:: python
+
+   >>> x.compute()
+   ZeroDivisionError(...)
+
+   >>> %pdb
+   >>> future = client.compute(x)
+   >>> client.recreate_error_locally(future)
+
+
+Remove Failed Futures Manually
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Sometimes only parts of your computations fail, for example if some rows of a
+CSV dataset are faulty in some way.  When running with the distributed
+scheduler you can remove chunks of your data that have produced bad results if
+you switch to dealing with Futures.
+
+.. code-block:: python
+
+   >>> import dask.dataframe as dd
+   >>> df = ...           # create dataframe
+   >>> df = df.persist()  # start computing on the cluster
+
+   >>> from distributed.client import futures_of
+   >>> futures = futures_of(df)  # get futures behind dataframe
+   >>> futures
+   [<Future: status: finished, type: pd.DataFrame, key: load-1>
+    <Future: status: finished, type: pd.DataFrame, key: load-2>
+    <Future: status: error, key: load-3>
+    <Future: status: pending, key: load-4>
+    <Future: status: error, key: load-5>]
+
+   >>> # wait until computation is done
+   >>> while any(f.status == 'pending' for f in futures):
+   ...     sleep(0.1)
+
+   >>> # pick out only the successful futures and reconstruct the dataframe
+   >>> good_futures = [f for f in futures if f.status == 'finished']
+   >>> df = dd.from_delayed(good_futures, meta=df._meta)
+
+This is a bit of a hack, but often practical when first exploring messy data.
+If you are using the concurrent.futures API (map, submit, gather) then this
+approach is more natural.
+
+
+Inspect Scheduling State
+------------------------
+
+Not all errors present themselves as Exceptions.  For example in a distributed
+system workers may die unexpectedly or your computation may be unreasonably
+slow due to inter-worker communication or scheduler overhead or one of several
+other issues.  Getting feedback about what's going on can help to identify
+both failures and general performance bottlenecks.
+
+For the single-machine scheduler see :doc:`diagnostics <diagnostics`
+documentation.  The rest of the section will assume that you are using the
+`distributed scheduler <http://distributed.readthedocs.io/en/latest/>`_ where
+these issues arise more commonly.
+
+Web Diagnostics
+~~~~~~~~~~~~~~~
+
+First, the distributed scheduler has a number of `diagnostic web pages
+<http://distributed.readthedocs.io/en/latest/web.html>`_ showing dozens of
+recorded metrics like CPU, memory, network, and disk use, a history of previous
+tasks, allocation of tasks to workers, worker memory pressure, work stealing,
+open file handle limits, etc..  *Many* problems can be correctly diagnosed by
+inspecting these pages.  By default these are available at
+``http://scheduler:8787/`` ``http://scheduler:8788/`` and ``http://worker:8789/``
+where ``scheduler`` and ``worker`` should be replaced by the addresses of the
+scheduler and each of the workers. See `web diagnostic docs
+<http://distributed.readthedocs.io/en/latest/web.html>`_ for more information.
+
+Logs
+~~~~
+
+The scheduler and workers and client all emits logs using `Python's standard
+logging module <https://docs.python.org/3/library/logging.html>`_.  By default
+these emit to standard error.  When Dask is launched by a cluster job scheduler
+(SGE/SLURM/YARN/Mesos/Marathon/Kubernetes/whatever) that system will track
+these logs and will have an interface to help you access them.  If you are
+launching Dask on your own they will probably dump to the screen unless you
+`redirect stderr to a file
+<https://en.wikipedia.org/wiki/Redirection_(computing)#Redirecting_to_and_from_the_standard_file_handles>`_
+.
+
+You can control the logging verbosity in the ``~/.dask/config.yaml`` file.
+Defaults currently look like the following:
+
+.. code-block:: yaml
+
+   logging:
+     distributed: info
+     distributed.client: warning
+     bokeh: error
+
+So for example you could add a line like ``distributed.worker: debug`` to get
+*very* verbose output from the workers.
+
+
+LocalCluster
+------------
+
+If you are using the distributed scheduler from a single machine you may be
+setting up workers manually using the command line interface or you may be
+using `LocalCluster <http://distributed.readthedocs.io/en/latest/local-cluster.html>`_
+which is what runs when you just call ``Client()``
+
+.. code-block:: python
+
+   >>> from dask.distributed import Client, LocalCluster
+   >>> client = Client()  # This is actually the following two commands
+
+   >>> cluster = LocalCluster()
+   >>> client = Client(cluster.scheduler.address)
+
+LocalCluster is useful because the scheduler and workers are in the same
+process with you, so you can easily inspect their `state
+<http://distributed.readthedocs.io/en/latest/scheduling-state.html>`_ while
+they run (they are running in a separate thread).
+
+.. code-block:: python
+
+   >>> cluster.scheduler.processing
+   {'worker-one:59858': {'inc-123', 'add-443'},
+    'worker-two:48248': {'inc-456'}}
+
+You can also do this for the workers *if* you run them without nanny processes.
+
+.. code-block:: python
+
+   >>> cluster = LocalCluster(nanny=False)
+   >>> client = Client(cluster)
+
+This can be very helpful if you want to use the dask.distributed API and still
+want to investigate what is going on directly within the workers.  Information
+is not distilled for you like it is in the web diagnostics, but you have full
+low-level access.
+
+
+Inspect state with IPython
+--------------------------
+
+Sometimes you want to inspect the state of your cluster, but you don't have the
+luxury of operating on a single machine.  In these cases you can launch an
+IPython kernel on the scheduler and on every worker, which lets you inspect
+state on the scheduler and workers as computations are completing.
+
+This does not give you the ability to run ``%pdb`` or ``%debug`` on remote
+machines, the tasks are still running in separate threads, and so are not
+easily accessible from an interactive IPython session.
+
+For more details, see the `Dask.distributed IPython docs
+<http://distributed.readthedocs.io/en/latest/ipython.html>`_.
diff --git a/docs/source/index.rst b/docs/source/index.rst
index b6bcb1770..5cc89e524 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -231,6 +231,7 @@ often a better choice.  If you are a *core developer*, then you should start her
 
 **Help & reference**
 
+* :doc:`debugging`
 * :doc:`support`
 * :doc:`changelog`
 * :doc:`presentations`
@@ -247,6 +248,7 @@ often a better choice.  If you are a *core developer*, then you should start her
    :hidden:
    :caption: Help & reference
 
+   debugging.rst
    support.rst
    changelog.rst
    presentations.rst


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index cb35eb3ac..b9b185b49 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -712,14 +712,31 @@ def test_ravel():
     assert_eq(np.ravel(x), da.ravel(a))
 
 
+def _maybe_len(l):
+    try:
+        return len(l)
+    except TypeError:
+        return 0
+
+
 @pytest.mark.parametrize('chunks', [(4, 6), (2, 6)])
-@pytest.mark.parametrize('shift', [3, 7, 9])
-@pytest.mark.parametrize('axis', [None, 0, 1, -1])
+@pytest.mark.parametrize('shift', [3, 7, 9, (3, 9), (7, 2)])
+@pytest.mark.parametrize('axis', [None, 0, 1, -1, (0, 1), (1, 0)])
 def test_roll(chunks, shift, axis):
     x = np.random.randint(10, size=(4, 6))
     a = from_array(x, chunks=chunks)
 
-    assert_eq(np.roll(x, shift, axis), da.roll(a, shift, axis))
+    if _maybe_len(shift) != _maybe_len(axis):
+        with pytest.raises(TypeError if axis is None else ValueError):
+            da.roll(a, shift, axis)
+    else:
+        if (_maybe_len(shift) > 1 and
+                LooseVersion(np.__version__) < LooseVersion("1.12.0")):
+            pytest.skip(
+                "NumPy %s doesn't support multiple axes with `roll`."
+                " Need NumPy 1.12.0 or greater." % np.__version__
+            )
+        assert_eq(np.roll(x, shift, axis), da.roll(a, shift, axis))
 
 
 @pytest.mark.parametrize('original_shape,new_shape,chunks', [
@@ -2572,6 +2589,44 @@ def test_repeat():
         assert all(concat(d.repeat(r).chunks))
 
 
+@pytest.mark.parametrize('shape, chunks', [
+    ((10,), (1,)),
+    ((10, 11, 13), (4, 5, 3)),
+])
+@pytest.mark.parametrize('reps', [0, 1, 2, 3, 5])
+def test_tile(shape, chunks, reps):
+    x = np.random.random(shape)
+    d = da.from_array(x, chunks=chunks)
+
+    assert_eq(np.tile(x, reps), da.tile(d, reps))
+
+
+@pytest.mark.parametrize('shape, chunks', [
+    ((10,), (1,)),
+    ((10, 11, 13), (4, 5, 3)),
+])
+@pytest.mark.parametrize('reps', [-1, -5])
+def test_tile_neg_reps(shape, chunks, reps):
+    x = np.random.random(shape)
+    d = da.from_array(x, chunks=chunks)
+
+    with pytest.raises(ValueError):
+        da.tile(d, reps)
+
+
+@pytest.mark.parametrize('shape, chunks', [
+    ((10,), (1,)),
+    ((10, 11, 13), (4, 5, 3)),
+])
+@pytest.mark.parametrize('reps', [[1], [1, 2]])
+def test_tile_array_reps(shape, chunks, reps):
+    x = np.random.random(shape)
+    d = da.from_array(x, chunks=chunks)
+
+    with pytest.raises(NotImplementedError):
+        da.tile(d, reps)
+
+
 def test_concatenate_stack_dont_warn():
     with warnings.catch_warnings(record=True) as record:
         da.concatenate([da.ones(2, chunks=1)] * 62)
@@ -2835,3 +2890,20 @@ def test_atop_with_numpy_arrays():
 
     s = da.sum(x)
     assert any(x is v for v in s.dask.values())
+
+
+@pytest.mark.parametrize('chunks', (100, 6))
+@pytest.mark.parametrize('other', [[0, 0, 1], [2, 1, 3], (0, 0, 1)])
+def test_elemwise_with_lists(chunks, other):
+    x = np.arange(12).reshape((4, 3))
+    d = da.arange(12, chunks=chunks).reshape((4, 3))
+
+    x2 = np.vstack([x[:, 0], x[:, 1], x[:, 2]]).T
+    d2 = da.vstack([d[:, 0], d[:, 1], d[:, 2]]).T
+
+    assert_eq(x2, d2)
+
+    x3 = x2 * other
+    d3 = d2 * other
+
+    assert_eq(x3, d3)
diff --git a/dask/tests/test_threaded.py b/dask/tests/test_threaded.py
index a377e9449..7b27a5963 100644
--- a/dask/tests/test_threaded.py
+++ b/dask/tests/test_threaded.py
@@ -1,6 +1,9 @@
-from multiprocessing.pool import ThreadPool
+import os
 import threading
-from threading import Thread
+import signal
+import subprocess
+import sys
+from multiprocessing.pool import ThreadPool
 from time import time, sleep
 
 import pytest
@@ -52,7 +55,7 @@ def test_threaded_within_thread():
     before = threading.active_count()
 
     for i in range(20):
-        t = Thread(target=f, args=(1,))
+        t = threading.Thread(target=f, args=(1,))
         t.daemon = True
         t.start()
         t.join()
@@ -91,7 +94,7 @@ def test_thread_safety():
 
     threads = []
     for i in range(20):
-        t = Thread(target=test_f)
+        t = threading.Thread(target=test_f)
         t.daemon = True
         t.start()
         threads.append(t)
@@ -100,3 +103,63 @@ def test_thread_safety():
         thread.join()
 
     assert L == [1] * 20
+
+
+code = """
+import sys
+from dask.threaded import get
+
+def signal_started():
+    sys.stdout.write('started\\n')
+    sys.stdout.flush()
+
+def long_task(x):
+    out = 0
+    N = 100000
+    for i in range(N):
+        for j in range(N):
+            out += 1
+
+dsk = {('x', i): (long_task, 'started') for i in range(100)}
+dsk['started'] = (signal_started,)
+dsk['x'] = (sum, list(dsk.keys()))
+get(dsk, 'x')
+"""
+
+
+# TODO: this test passes locally on windows, but fails on appveyor
+# because the ctrl-c event also tears down their infrastructure.
+# There's probably a better way to test this, but for now we'll mark
+# it slow (slow tests are skipped on appveyor).
+@pytest.mark.slow
+def test_interrupt():
+    try:
+        proc = subprocess.Popen([sys.executable, '-c', code],
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE)
+        # Wait for scheduler to start
+        msg = proc.stdout.readline()
+        if msg != 'started\n' and proc.poll() is not None:
+            assert False, "subprocess failed"
+        # Scheduler has started, send an interrupt
+        sigint = signal.CTRL_C_EVENT if os.name == 'nt' else signal.SIGINT
+        try:
+            proc.send_signal(sigint)
+            # Wait a bit for it to die
+            start = time()
+            while time() - start < 0.5:
+                if proc.poll() is not None:
+                    break
+                sleep(0.05)
+            else:
+                assert False, "KeyboardInterrupt Failed"
+        except KeyboardInterrupt:
+            # On windows the interrupt is also raised in this process.
+            # That's silly, ignore it.
+            pass
+        # Ensure KeyboardInterrupt in traceback
+        stderr = proc.stderr.read()
+        assert "KeyboardInterrupt" in stderr.decode()
+    except:
+        proc.terminate()
+        raise