We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Wait/sync doesn't break on KeyboardInterrupt
Calling a function like `wait` that uses `sync` can not be broken with `KeyboardInterrupt`.  Two questions
1.  How does one test keyboard interrupts?  Is there a way to simulate the signal?
2.  What could be happening here?  We are waiting on a threading `Event` when the `KeyboardInterrupt` _should_ interrupt us.  https://github.com/blaze/distributed/blob/master/distributed/utils.py#L104


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/utils.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from collections import Iterable
4 from contextlib import contextmanager
5 import logging
6 import os
7 import re
8 import socket
9 import sys
10 import tempfile
11 import traceback
12 
13 from dask import istask
14 from toolz import memoize
15 from tornado import gen
16 
17 
18 logger = logging.getLogger(__name__)
19 
20 
21 def funcname(func):
22     """Get the name of a function."""
23     while hasattr(func, 'func'):
24         func = func.func
25     try:
26         return func.__name__
27     except:
28         return str(func)
29 
30 
31 def get_ip():
32     return [(s.connect(('8.8.8.8', 80)), s.getsockname()[0], s.close())
33         for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]
34 
35 
36 
37 @contextmanager
38 def ignoring(*exceptions):
39     try:
40         yield
41     except exceptions:
42         pass
43 
44 
45 @gen.coroutine
46 def ignore_exceptions(coroutines, *exceptions):
47     """ Process list of coroutines, ignoring certain exceptions
48 
49     >>> coroutines = [cor(...) for ...]  # doctest: +SKIP
50     >>> x = yield ignore_exceptions(coroutines, TypeError)  # doctest: +SKIP
51     """
52     wait_iterator = gen.WaitIterator(*coroutines)
53     results = []
54     while not wait_iterator.done():
55         with ignoring(*exceptions):
56             result = yield wait_iterator.next()
57             results.append(result)
58     raise gen.Return(results)
59 
60 
61 @gen.coroutine
62 def All(*args):
63     """ Wait on many tasks at the same time
64 
65     Err once any of the tasks err.
66 
67     See https://github.com/tornadoweb/tornado/issues/1546
68     """
69     if len(args) == 1 and isinstance(args[0], Iterable):
70         args = args[0]
71     tasks = gen.WaitIterator(*args)
72     results = [None for _ in args]
73     while not tasks.done():
74         result = yield tasks.next()
75         results[tasks.current_index] = result
76     raise gen.Return(results)
77 
78 
79 def sync(loop, func, *args, **kwargs):
80     """ Run coroutine in loop running in separate thread """
81     if not loop._running:
82         try:
83             return loop.run_sync(lambda: func(*args, **kwargs))
84         except RuntimeError:  # loop already running
85             pass
86 
87     from threading import Event
88     e = Event()
89     result = [None]
90     error = [False]
91 
92     @gen.coroutine
93     def f():
94         try:
95             result[0] = yield gen.maybe_future(func(*args, **kwargs))
96         except Exception as exc:
97             logger.exception(exc)
98             result[0] = exc
99             error[0] = True
100         finally:
101             e.set()
102 
103     a = loop.add_callback(f)
104     e.wait()
105     if error[0]:
106         raise result[0]
107     else:
108         return result[0]
109 
110 
111 @contextmanager
112 def tmp_text(filename, text):
113     fn = os.path.join(tempfile.gettempdir(), filename)
114     with open(fn, 'w') as f:
115         f.write(text)
116 
117     try:
118         yield fn
119     finally:
120         if os.path.exists(fn):
121             os.remove(fn)
122 
123 
124 def clear_queue(q):
125     while not q.empty():
126         q.get_nowait()
127 
128 
129 def is_kernel():
130     """ Determine if we're running within an IPython kernel
131 
132     >>> is_kernel()
133     False
134     """
135     # http://stackoverflow.com/questions/34091701/determine-if-were-in-an-ipython-notebook-session
136     if 'IPython' not in sys.modules:  # IPython hasn't been imported
137         return False
138     from IPython import get_ipython
139     # check for `kernel` attribute on the IPython instance
140     return getattr(get_ipython(), 'kernel', None) is not None
141 
142 
143 def _deps(dsk, arg):
144     """ Get dependencies from keys or tasks
145 
146     Helper function for get_dependencies.
147 
148     Examples
149     --------
150     >>> inc = lambda x: x + 1
151     >>> add = lambda x, y: x + y
152 
153     >>> dsk = {'x': 1, 'y': 2}
154 
155     >>> _deps(dsk, 'x')
156     ['x']
157     >>> _deps(dsk, (add, 'x', 1))
158     ['x']
159     >>> _deps(dsk, ['x', 'y'])
160     ['x', 'y']
161     >>> _deps(dsk, {'name': 'x'})
162     ['x']
163     >>> _deps(dsk, (add, 'x', (inc, 'y')))  # doctest: +SKIP
164     ['x', 'y']
165     """
166     if istask(arg):
167         result = []
168         for a in arg[1:]:
169             result.extend(_deps(dsk, a))
170         return result
171     if isinstance(arg, list):
172         return sum([_deps(dsk, a) for a in arg], [])
173     if isinstance(arg, dict):
174         return sum([_deps(dsk, v) for v in arg.values()], [])
175     try:
176         if arg not in dsk:
177             return []
178     except TypeError:  # not hashable
179             return []
180     return [arg]
181 
182 
183 def key_split(s):
184     """
185     >>> key_split('x-1')
186     'x'
187     >>> key_split('x-1-2-3')
188     'x'
189     >>> key_split(('x-2', 1))
190     'x'
191     >>> key_split(None)
192     'Other'
193     """
194     if isinstance(s, tuple):
195         return key_split(s[0])
196     try:
197         return s.split('-', 1)[0]
198     except:
199         return 'Other'
200 
201 
202 @contextmanager
203 def log_errors():
204     try:
205         yield
206     except gen.Return:
207         raise
208     except Exception as e:
209         logger.exception(e)
210         raise
211 
212 
213 @memoize
214 def ensure_ip(hostname):
215     """ Ensure that address is an IP address
216 
217     >>> ensure_ip('localhost')
218     '127.0.0.1'
219     >>> ensure_ip('123.123.123.123')  # pass through IP addresses
220     '123.123.123.123'
221     """
222     if re.match('\d+\.\d+\.\d+\.\d+', hostname):  # is IP
223         return hostname
224     else:
225         return socket.gethostbyname(hostname)
226 
227 
228 def get_traceback():
229     exc_type, exc_value, exc_traceback = sys.exc_info()
230     tb = traceback.format_tb(exc_traceback)
231     tb = [line[:10000] for line in tb]
232     return tb
233 
234 
235 def truncate_exception(e, n=10000):
236     """ Truncate exception to be about a certain length """
237     if len(str(e)) > n:
238         try:
239             return type(e)("Long error message",
240                            str(e)[:n])
241         except:
242             return Exception("Long error message",
243                               type(e),
244                               str(e)[:n])
245     else:
246         return e
247 
248 
249 import logging
250 logging.basicConfig(format='%(name)s - %(levelname)s - %(message)s',
251                     level=logging.INFO)
252 
253 # http://stackoverflow.com/questions/21234772/python-tornado-disable-logging-to-stderr
254 stream = logging.StreamHandler(sys.stderr)
255 stream.setLevel(logging.CRITICAL)
256 logging.getLogger('tornado').addHandler(stream)
257 logging.getLogger('tornado').propagate = False
258
```
--- END FILES ---

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

Here is an example:

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

```

diff --git a/distributed/utils.py b/distributed/utils.py
--- a/distributed/utils.py
+++ b/distributed/utils.py
@@ -101,7 +101,8 @@
             e.set()
 
     a = loop.add_callback(f)
-    e.wait()
+    while not e.is_set():
+        e.wait(1000000)
     if error[0]:
         raise result[0]
     else:

{"golden_diff": "diff --git a/distributed/utils.py b/distributed/utils.py\n--- a/distributed/utils.py\n+++ b/distributed/utils.py\n@@ -101,7 +101,8 @@\n             e.set()\n \n     a = loop.add_callback(f)\n-    e.wait()\n+    while not e.is_set():\n+        e.wait(1000000)\n     if error[0]:\n         raise result[0]\n     else:\n", "issue": "Wait/sync doesn't break on KeyboardInterrupt\nCalling a function like `wait` that uses `sync` can not be broken with `KeyboardInterrupt`.  Two questions\n1.  How does one test keyboard interrupts?  Is there a way to simulate the signal?\n2.  What could be happening here?  We are waiting on a threading `Event` when the `KeyboardInterrupt` _should_ interrupt us.  https://github.com/blaze/distributed/blob/master/distributed/utils.py#L104\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom collections import Iterable\nfrom contextlib import contextmanager\nimport logging\nimport os\nimport re\nimport socket\nimport sys\nimport tempfile\nimport traceback\n\nfrom dask import istask\nfrom toolz import memoize\nfrom tornado import gen\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef funcname(func):\n    \"\"\"Get the name of a function.\"\"\"\n    while hasattr(func, 'func'):\n        func = func.func\n    try:\n        return func.__name__\n    except:\n        return str(func)\n\n\ndef get_ip():\n    return [(s.connect(('8.8.8.8', 80)), s.getsockname()[0], s.close())\n        for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]\n\n\n\n@contextmanager\ndef ignoring(*exceptions):\n    try:\n        yield\n    except exceptions:\n        pass\n\n\n@gen.coroutine\ndef ignore_exceptions(coroutines, *exceptions):\n    \"\"\" Process list of coroutines, ignoring certain exceptions\n\n    >>> coroutines = [cor(...) for ...]  # doctest: +SKIP\n    >>> x = yield ignore_exceptions(coroutines, TypeError)  # doctest: +SKIP\n    \"\"\"\n    wait_iterator = gen.WaitIterator(*coroutines)\n    results = []\n    while not wait_iterator.done():\n        with ignoring(*exceptions):\n            result = yield wait_iterator.next()\n            results.append(result)\n    raise gen.Return(results)\n\n\n@gen.coroutine\ndef All(*args):\n    \"\"\" Wait on many tasks at the same time\n\n    Err once any of the tasks err.\n\n    See https://github.com/tornadoweb/tornado/issues/1546\n    \"\"\"\n    if len(args) == 1 and isinstance(args[0], Iterable):\n        args = args[0]\n    tasks = gen.WaitIterator(*args)\n    results = [None for _ in args]\n    while not tasks.done():\n        result = yield tasks.next()\n        results[tasks.current_index] = result\n    raise gen.Return(results)\n\n\ndef sync(loop, func, *args, **kwargs):\n    \"\"\" Run coroutine in loop running in separate thread \"\"\"\n    if not loop._running:\n        try:\n            return loop.run_sync(lambda: func(*args, **kwargs))\n        except RuntimeError:  # loop already running\n            pass\n\n    from threading import Event\n    e = Event()\n    result = [None]\n    error = [False]\n\n    @gen.coroutine\n    def f():\n        try:\n            result[0] = yield gen.maybe_future(func(*args, **kwargs))\n        except Exception as exc:\n            logger.exception(exc)\n            result[0] = exc\n            error[0] = True\n        finally:\n            e.set()\n\n    a = loop.add_callback(f)\n    e.wait()\n    if error[0]:\n        raise result[0]\n    else:\n        return result[0]\n\n\n@contextmanager\ndef tmp_text(filename, text):\n    fn = os.path.join(tempfile.gettempdir(), filename)\n    with open(fn, 'w') as f:\n        f.write(text)\n\n    try:\n        yield fn\n    finally:\n        if os.path.exists(fn):\n            os.remove(fn)\n\n\ndef clear_queue(q):\n    while not q.empty():\n        q.get_nowait()\n\n\ndef is_kernel():\n    \"\"\" Determine if we're running within an IPython kernel\n\n    >>> is_kernel()\n    False\n    \"\"\"\n    # http://stackoverflow.com/questions/34091701/determine-if-were-in-an-ipython-notebook-session\n    if 'IPython' not in sys.modules:  # IPython hasn't been imported\n        return False\n    from IPython import get_ipython\n    # check for `kernel` attribute on the IPython instance\n    return getattr(get_ipython(), 'kernel', None) is not None\n\n\ndef _deps(dsk, arg):\n    \"\"\" Get dependencies from keys or tasks\n\n    Helper function for get_dependencies.\n\n    Examples\n    --------\n    >>> inc = lambda x: x + 1\n    >>> add = lambda x, y: x + y\n\n    >>> dsk = {'x': 1, 'y': 2}\n\n    >>> _deps(dsk, 'x')\n    ['x']\n    >>> _deps(dsk, (add, 'x', 1))\n    ['x']\n    >>> _deps(dsk, ['x', 'y'])\n    ['x', 'y']\n    >>> _deps(dsk, {'name': 'x'})\n    ['x']\n    >>> _deps(dsk, (add, 'x', (inc, 'y')))  # doctest: +SKIP\n    ['x', 'y']\n    \"\"\"\n    if istask(arg):\n        result = []\n        for a in arg[1:]:\n            result.extend(_deps(dsk, a))\n        return result\n    if isinstance(arg, list):\n        return sum([_deps(dsk, a) for a in arg], [])\n    if isinstance(arg, dict):\n        return sum([_deps(dsk, v) for v in arg.values()], [])\n    try:\n        if arg not in dsk:\n            return []\n    except TypeError:  # not hashable\n            return []\n    return [arg]\n\n\ndef key_split(s):\n    \"\"\"\n    >>> key_split('x-1')\n    'x'\n    >>> key_split('x-1-2-3')\n    'x'\n    >>> key_split(('x-2', 1))\n    'x'\n    >>> key_split(None)\n    'Other'\n    \"\"\"\n    if isinstance(s, tuple):\n        return key_split(s[0])\n    try:\n        return s.split('-', 1)[0]\n    except:\n        return 'Other'\n\n\n@contextmanager\ndef log_errors():\n    try:\n        yield\n    except gen.Return:\n        raise\n    except Exception as e:\n        logger.exception(e)\n        raise\n\n\n@memoize\ndef ensure_ip(hostname):\n    \"\"\" Ensure that address is an IP address\n\n    >>> ensure_ip('localhost')\n    '127.0.0.1'\n    >>> ensure_ip('123.123.123.123')  # pass through IP addresses\n    '123.123.123.123'\n    \"\"\"\n    if re.match('\\d+\\.\\d+\\.\\d+\\.\\d+', hostname):  # is IP\n        return hostname\n    else:\n        return socket.gethostbyname(hostname)\n\n\ndef get_traceback():\n    exc_type, exc_value, exc_traceback = sys.exc_info()\n    tb = traceback.format_tb(exc_traceback)\n    tb = [line[:10000] for line in tb]\n    return tb\n\n\ndef truncate_exception(e, n=10000):\n    \"\"\" Truncate exception to be about a certain length \"\"\"\n    if len(str(e)) > n:\n        try:\n            return type(e)(\"Long error message\",\n                           str(e)[:n])\n        except:\n            return Exception(\"Long error message\",\n                              type(e),\n                              str(e)[:n])\n    else:\n        return e\n\n\nimport logging\nlogging.basicConfig(format='%(name)s - %(levelname)s - %(message)s',\n                    level=logging.INFO)\n\n# http://stackoverflow.com/questions/21234772/python-tornado-disable-logging-to-stderr\nstream = logging.StreamHandler(sys.stderr)\nstream.setLevel(logging.CRITICAL)\nlogging.getLogger('tornado').addHandler(stream)\nlogging.getLogger('tornado').propagate = False\n", "path": "distributed/utils.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom collections import Iterable\nfrom contextlib import contextmanager\nimport logging\nimport os\nimport re\nimport socket\nimport sys\nimport tempfile\nimport traceback\n\nfrom dask import istask\nfrom toolz import memoize\nfrom tornado import gen\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef funcname(func):\n    \"\"\"Get the name of a function.\"\"\"\n    while hasattr(func, 'func'):\n        func = func.func\n    try:\n        return func.__name__\n    except:\n        return str(func)\n\n\ndef get_ip():\n    return [(s.connect(('8.8.8.8', 80)), s.getsockname()[0], s.close())\n        for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]\n\n\n\n@contextmanager\ndef ignoring(*exceptions):\n    try:\n        yield\n    except exceptions:\n        pass\n\n\n@gen.coroutine\ndef ignore_exceptions(coroutines, *exceptions):\n    \"\"\" Process list of coroutines, ignoring certain exceptions\n\n    >>> coroutines = [cor(...) for ...]  # doctest: +SKIP\n    >>> x = yield ignore_exceptions(coroutines, TypeError)  # doctest: +SKIP\n    \"\"\"\n    wait_iterator = gen.WaitIterator(*coroutines)\n    results = []\n    while not wait_iterator.done():\n        with ignoring(*exceptions):\n            result = yield wait_iterator.next()\n            results.append(result)\n    raise gen.Return(results)\n\n\n@gen.coroutine\ndef All(*args):\n    \"\"\" Wait on many tasks at the same time\n\n    Err once any of the tasks err.\n\n    See https://github.com/tornadoweb/tornado/issues/1546\n    \"\"\"\n    if len(args) == 1 and isinstance(args[0], Iterable):\n        args = args[0]\n    tasks = gen.WaitIterator(*args)\n    results = [None for _ in args]\n    while not tasks.done():\n        result = yield tasks.next()\n        results[tasks.current_index] = result\n    raise gen.Return(results)\n\n\ndef sync(loop, func, *args, **kwargs):\n    \"\"\" Run coroutine in loop running in separate thread \"\"\"\n    if not loop._running:\n        try:\n            return loop.run_sync(lambda: func(*args, **kwargs))\n        except RuntimeError:  # loop already running\n            pass\n\n    from threading import Event\n    e = Event()\n    result = [None]\n    error = [False]\n\n    @gen.coroutine\n    def f():\n        try:\n            result[0] = yield gen.maybe_future(func(*args, **kwargs))\n        except Exception as exc:\n            logger.exception(exc)\n            result[0] = exc\n            error[0] = True\n        finally:\n            e.set()\n\n    a = loop.add_callback(f)\n    while not e.is_set():\n        e.wait(1000000)\n    if error[0]:\n        raise result[0]\n    else:\n        return result[0]\n\n\n@contextmanager\ndef tmp_text(filename, text):\n    fn = os.path.join(tempfile.gettempdir(), filename)\n    with open(fn, 'w') as f:\n        f.write(text)\n\n    try:\n        yield fn\n    finally:\n        if os.path.exists(fn):\n            os.remove(fn)\n\n\ndef clear_queue(q):\n    while not q.empty():\n        q.get_nowait()\n\n\ndef is_kernel():\n    \"\"\" Determine if we're running within an IPython kernel\n\n    >>> is_kernel()\n    False\n    \"\"\"\n    # http://stackoverflow.com/questions/34091701/determine-if-were-in-an-ipython-notebook-session\n    if 'IPython' not in sys.modules:  # IPython hasn't been imported\n        return False\n    from IPython import get_ipython\n    # check for `kernel` attribute on the IPython instance\n    return getattr(get_ipython(), 'kernel', None) is not None\n\n\ndef _deps(dsk, arg):\n    \"\"\" Get dependencies from keys or tasks\n\n    Helper function for get_dependencies.\n\n    Examples\n    --------\n    >>> inc = lambda x: x + 1\n    >>> add = lambda x, y: x + y\n\n    >>> dsk = {'x': 1, 'y': 2}\n\n    >>> _deps(dsk, 'x')\n    ['x']\n    >>> _deps(dsk, (add, 'x', 1))\n    ['x']\n    >>> _deps(dsk, ['x', 'y'])\n    ['x', 'y']\n    >>> _deps(dsk, {'name': 'x'})\n    ['x']\n    >>> _deps(dsk, (add, 'x', (inc, 'y')))  # doctest: +SKIP\n    ['x', 'y']\n    \"\"\"\n    if istask(arg):\n        result = []\n        for a in arg[1:]:\n            result.extend(_deps(dsk, a))\n        return result\n    if isinstance(arg, list):\n        return sum([_deps(dsk, a) for a in arg], [])\n    if isinstance(arg, dict):\n        return sum([_deps(dsk, v) for v in arg.values()], [])\n    try:\n        if arg not in dsk:\n            return []\n    except TypeError:  # not hashable\n            return []\n    return [arg]\n\n\ndef key_split(s):\n    \"\"\"\n    >>> key_split('x-1')\n    'x'\n    >>> key_split('x-1-2-3')\n    'x'\n    >>> key_split(('x-2', 1))\n    'x'\n    >>> key_split(None)\n    'Other'\n    \"\"\"\n    if isinstance(s, tuple):\n        return key_split(s[0])\n    try:\n        return s.split('-', 1)[0]\n    except:\n        return 'Other'\n\n\n@contextmanager\ndef log_errors():\n    try:\n        yield\n    except gen.Return:\n        raise\n    except Exception as e:\n        logger.exception(e)\n        raise\n\n\n@memoize\ndef ensure_ip(hostname):\n    \"\"\" Ensure that address is an IP address\n\n    >>> ensure_ip('localhost')\n    '127.0.0.1'\n    >>> ensure_ip('123.123.123.123')  # pass through IP addresses\n    '123.123.123.123'\n    \"\"\"\n    if re.match('\\d+\\.\\d+\\.\\d+\\.\\d+', hostname):  # is IP\n        return hostname\n    else:\n        return socket.gethostbyname(hostname)\n\n\ndef get_traceback():\n    exc_type, exc_value, exc_traceback = sys.exc_info()\n    tb = traceback.format_tb(exc_traceback)\n    tb = [line[:10000] for line in tb]\n    return tb\n\n\ndef truncate_exception(e, n=10000):\n    \"\"\" Truncate exception to be about a certain length \"\"\"\n    if len(str(e)) > n:\n        try:\n            return type(e)(\"Long error message\",\n                           str(e)[:n])\n        except:\n            return Exception(\"Long error message\",\n                              type(e),\n                              str(e)[:n])\n    else:\n        return e\n\n\nimport logging\nlogging.basicConfig(format='%(name)s - %(levelname)s - %(message)s',\n                    level=logging.INFO)\n\n# http://stackoverflow.com/questions/21234772/python-tornado-disable-logging-to-stderr\nstream = logging.StreamHandler(sys.stderr)\nstream.setLevel(logging.CRITICAL)\nlogging.getLogger('tornado').addHandler(stream)\nlogging.getLogger('tornado').propagate = False\n", "path": "distributed/utils.py"}]}