We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Accessing `ipython` on workers through `client.start_ipython_workers()`
I am trying to see if I can run ipython on workers interactively. The first example in the documentation (https://distributed.dask.org/en/latest/ipython.html) works, while the second one is not working for me. Here's the minimal code to reproduce:

```python
from dask.distributed import Client
client = Client()

info = client.start_ipython_workers(magic_names="w*")
%w0 1+1
# this gives TimeoutError: Timeout waiting for IPython output
```

This is more of a curiosity for me, just trying to explore what is possible. I posted it also as a question on StackOverflow: https://stackoverflow.com/questions/66167619/how-to-use-client-start-ipython-workers-in-dask-distributed

**Environment**:

- Dask version: 2021.02.0
- Python version: Python 3.7.9
- Operating System: MacOS Catalina
- Install method (conda, pip, source): conda


--- 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/_ipython_utils.py`
Content:
```
1 """Utilities for integrating with IPython
2 
3 These functions should probably reside in Jupyter and IPython repositories,
4 after which we can import them instead of having our own definitions.
5 """
6 
7 import atexit
8 import os
9 
10 try:
11     import queue
12 except ImportError:
13     # Python 2
14     import Queue as queue
15 from subprocess import Popen
16 import sys
17 from threading import Thread
18 from uuid import uuid4
19 
20 from tornado.gen import TimeoutError
21 from tornado.ioloop import IOLoop
22 from threading import Event
23 
24 from IPython import get_ipython
25 from jupyter_client import BlockingKernelClient, write_connection_file
26 from jupyter_core.paths import jupyter_runtime_dir
27 
28 
29 OUTPUT_TIMEOUT = 10
30 
31 
32 def run_cell_remote(ip, kc, cell):
33     """Run a cell on a KernelClient
34 
35     Any output from the cell will be redisplayed in the local session.
36     """
37     msg_id = kc.execute(cell)
38 
39     in_kernel = getattr(ip, "kernel", False)
40     if in_kernel:
41         socket = ip.display_pub.pub_socket
42         session = ip.display_pub.session
43         parent_header = ip.display_pub.parent_header
44 
45     while True:
46         try:
47             msg = kc.get_iopub_msg(timeout=OUTPUT_TIMEOUT)
48         except queue.Empty:
49             raise TimeoutError("Timeout waiting for IPython output")
50 
51         if msg["parent_header"].get("msg_id") != msg_id:
52             continue
53         msg_type = msg["header"]["msg_type"]
54         content = msg["content"]
55         if msg_type == "status":
56             if content["execution_state"] == "idle":
57                 # idle means output is done
58                 break
59         elif msg_type == "stream":
60             stream = getattr(sys, content["name"])
61             stream.write(content["text"])
62         elif msg_type in ("display_data", "execute_result", "error"):
63             if in_kernel:
64                 session.send(socket, msg_type, content, parent=parent_header)
65             else:
66                 if msg_type == "error":
67                     print("\n".join(content["traceback"]), file=sys.stderr)
68                 else:
69                     sys.stdout.write(content["data"].get("text/plain", ""))
70         else:
71             pass
72 
73 
74 def register_worker_magic(connection_info, magic_name="worker"):
75     """Register a %worker magic, given connection_info.
76 
77     Both a line and cell magic are registered,
78     which run the given cell in a remote kernel.
79     """
80     ip = get_ipython()
81     info = dict(connection_info)  # copy
82     key = info.pop("key")
83     kc = BlockingKernelClient(**connection_info)
84     kc.session.key = key
85     kc.start_channels()
86 
87     def remote(line, cell=None):
88         """Run the current cell on a remote IPython kernel"""
89         if cell is None:
90             # both line and cell magic
91             cell = line
92         run_cell_remote(ip, kc, cell)
93 
94     remote.client = kc  # preserve reference on kc, largely for mocking
95     ip.register_magic_function(remote, magic_kind="line", magic_name=magic_name)
96     ip.register_magic_function(remote, magic_kind="cell", magic_name=magic_name)
97 
98 
99 def remote_magic(line, cell=None):
100     """A magic for running code on a specified remote worker
101 
102     The connection_info dict of the worker will be looked up
103     as the first positional arg to the magic.
104     The rest of the line (or the entire cell for a %%cell magic)
105     will be passed to the remote kernel.
106 
107     Usage:
108 
109         info = e.start_ipython(worker)[worker]
110         %remote info print(worker.data)
111     """
112     # get connection info from IPython's user namespace
113     ip = get_ipython()
114     split_line = line.split(None, 1)
115     info_name = split_line[0]
116     if info_name not in ip.user_ns:
117         raise NameError(info_name)
118     connection_info = dict(ip.user_ns[info_name])
119 
120     if not cell:  # line magic, use the rest of the line
121         if len(split_line) == 1:
122             raise ValueError("I need some code to run!")
123         cell = split_line[1]
124 
125     # turn info dict to hashable str for use as lookup key in _clients cache
126     key = ",".join(map(str, sorted(connection_info.items())))
127     session_key = connection_info.pop("key")
128 
129     if key in remote_magic._clients:
130         kc = remote_magic._clients[key]
131     else:
132         kc = BlockingKernelClient(**connection_info)
133         kc.session.key = session_key
134         kc.start_channels()
135         kc.wait_for_ready(timeout=10)
136         remote_magic._clients[key] = kc
137 
138     # actually run the code
139     run_cell_remote(ip, kc, cell)
140 
141 
142 # cache clients for re-use in remote magic
143 remote_magic._clients = {}
144 
145 
146 def register_remote_magic(magic_name="remote"):
147     """Define the parameterized %remote magic
148 
149     See remote_magic above for details.
150     """
151     ip = get_ipython()
152     if ip is None:
153         return  # do nothing if IPython's not running
154     ip.register_magic_function(remote_magic, magic_kind="line", magic_name=magic_name)
155     ip.register_magic_function(remote_magic, magic_kind="cell", magic_name=magic_name)
156 
157 
158 def connect_qtconsole(connection_info, name=None, extra_args=None):
159     """Open a QtConsole connected to a worker who has the given future
160 
161     - identify worker with who_has
162     - start IPython kernel on the worker
163     - start qtconsole connected to the kernel
164     """
165     runtime_dir = jupyter_runtime_dir()
166     if name is None:
167         name = uuid4().hex
168 
169     path = os.path.join(runtime_dir, name + ".json")
170     write_connection_file(path, **connection_info)
171     cmd = ["jupyter", "qtconsole", "--existing", path]
172     if extra_args:
173         cmd.extend(extra_args)
174     Popen(cmd)
175 
176     @atexit.register
177     def _cleanup_connection_file():
178         """Cleanup our connection file when we exit."""
179         try:
180             os.remove(path)
181         except OSError:
182             pass
183 
184 
185 def start_ipython(ip=None, ns=None, log=None):
186     """Start an IPython kernel in a thread
187 
188     Parameters
189     ----------
190     ip : str
191         The IP address to listen on (likely the parent object's ip).
192     ns : dict
193         Any names that should be injected into the IPython namespace.
194     log : logger instance
195         Hook up IPython's logging to an existing logger instead of the default.
196     """
197     from IPython import get_ipython
198 
199     if get_ipython() is not None:
200         raise RuntimeError("Cannot start IPython, it's already running.")
201 
202     from zmq.eventloop.ioloop import ZMQIOLoop
203     from ipykernel.kernelapp import IPKernelApp
204 
205     # save the global IOLoop instance
206     # since IPython relies on it, but we are going to put it in a thread.
207     save_inst = IOLoop.instance()
208     IOLoop.clear_instance()
209     zmq_loop = ZMQIOLoop()
210     zmq_loop.install()
211 
212     # start IPython, disabling its signal handlers that won't work due to running in a thread:
213     app = IPKernelApp.instance(log=log)
214     # Don't connect to the history database
215     app.config.HistoryManager.hist_file = ":memory:"
216     # listen on all interfaces, so remote clients can connect:
217     if ip:
218         app.ip = ip
219     # disable some signal handling, logging
220 
221     def noop():
222         return None
223 
224     app.init_signal = noop
225     app.log_connection_info = noop
226 
227     # start IPython in a thread
228     # initialization happens in the thread to avoid threading problems
229     # with the sqlite history
230     evt = Event()
231 
232     def _start():
233         app.initialize([])
234         app.kernel.pre_handler_hook = noop
235         app.kernel.post_handler_hook = noop
236         app.kernel.start()
237         app.kernel.loop = IOLoop.instance()
238         # save self in the IPython namespace as 'worker'
239         # inject things into the IPython namespace
240         if ns:
241             app.kernel.shell.user_ns.update(ns)
242         evt.set()
243         zmq_loop.start()
244 
245     zmq_loop_thread = Thread(target=_start)
246     zmq_loop_thread.daemon = True
247     zmq_loop_thread.start()
248     assert evt.wait(timeout=5), "IPython didn't start in a reasonable amount of time."
249 
250     # put the global IOLoop instance back:
251     IOLoop.clear_instance()
252     save_inst.install()
253     return app
254
```
--- 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/_ipython_utils.py b/distributed/_ipython_utils.py
--- a/distributed/_ipython_utils.py
+++ b/distributed/_ipython_utils.py
@@ -199,16 +199,8 @@
     if get_ipython() is not None:
         raise RuntimeError("Cannot start IPython, it's already running.")
 
-    from zmq.eventloop.ioloop import ZMQIOLoop
     from ipykernel.kernelapp import IPKernelApp
 
-    # save the global IOLoop instance
-    # since IPython relies on it, but we are going to put it in a thread.
-    save_inst = IOLoop.instance()
-    IOLoop.clear_instance()
-    zmq_loop = ZMQIOLoop()
-    zmq_loop.install()
-
     # start IPython, disabling its signal handlers that won't work due to running in a thread:
     app = IPKernelApp.instance(log=log)
     # Don't connect to the history database
@@ -234,20 +226,17 @@
         app.kernel.pre_handler_hook = noop
         app.kernel.post_handler_hook = noop
         app.kernel.start()
-        app.kernel.loop = IOLoop.instance()
         # save self in the IPython namespace as 'worker'
         # inject things into the IPython namespace
         if ns:
             app.kernel.shell.user_ns.update(ns)
         evt.set()
-        zmq_loop.start()
+        # start the app's IOLoop in its thread
+        IOLoop.current().start()
 
     zmq_loop_thread = Thread(target=_start)
     zmq_loop_thread.daemon = True
     zmq_loop_thread.start()
     assert evt.wait(timeout=5), "IPython didn't start in a reasonable amount of time."
 
-    # put the global IOLoop instance back:
-    IOLoop.clear_instance()
-    save_inst.install()
     return app

{"golden_diff": "diff --git a/distributed/_ipython_utils.py b/distributed/_ipython_utils.py\n--- a/distributed/_ipython_utils.py\n+++ b/distributed/_ipython_utils.py\n@@ -199,16 +199,8 @@\n     if get_ipython() is not None:\n         raise RuntimeError(\"Cannot start IPython, it's already running.\")\n \n-    from zmq.eventloop.ioloop import ZMQIOLoop\n     from ipykernel.kernelapp import IPKernelApp\n \n-    # save the global IOLoop instance\n-    # since IPython relies on it, but we are going to put it in a thread.\n-    save_inst = IOLoop.instance()\n-    IOLoop.clear_instance()\n-    zmq_loop = ZMQIOLoop()\n-    zmq_loop.install()\n-\n     # start IPython, disabling its signal handlers that won't work due to running in a thread:\n     app = IPKernelApp.instance(log=log)\n     # Don't connect to the history database\n@@ -234,20 +226,17 @@\n         app.kernel.pre_handler_hook = noop\n         app.kernel.post_handler_hook = noop\n         app.kernel.start()\n-        app.kernel.loop = IOLoop.instance()\n         # save self in the IPython namespace as 'worker'\n         # inject things into the IPython namespace\n         if ns:\n             app.kernel.shell.user_ns.update(ns)\n         evt.set()\n-        zmq_loop.start()\n+        # start the app's IOLoop in its thread\n+        IOLoop.current().start()\n \n     zmq_loop_thread = Thread(target=_start)\n     zmq_loop_thread.daemon = True\n     zmq_loop_thread.start()\n     assert evt.wait(timeout=5), \"IPython didn't start in a reasonable amount of time.\"\n \n-    # put the global IOLoop instance back:\n-    IOLoop.clear_instance()\n-    save_inst.install()\n     return app\n", "issue": "Accessing `ipython` on workers through `client.start_ipython_workers()`\nI am trying to see if I can run ipython on workers interactively. The first example in the documentation (https://distributed.dask.org/en/latest/ipython.html) works, while the second one is not working for me. Here's the minimal code to reproduce:\r\n\r\n```python\r\nfrom dask.distributed import Client\r\nclient = Client()\r\n\r\ninfo = client.start_ipython_workers(magic_names=\"w*\")\r\n%w0 1+1\r\n# this gives TimeoutError: Timeout waiting for IPython output\r\n```\r\n\r\nThis is more of a curiosity for me, just trying to explore what is possible. I posted it also as a question on StackOverflow: https://stackoverflow.com/questions/66167619/how-to-use-client-start-ipython-workers-in-dask-distributed\r\n\r\n**Environment**:\r\n\r\n- Dask version: 2021.02.0\r\n- Python version: Python 3.7.9\r\n- Operating System: MacOS Catalina\r\n- Install method (conda, pip, source): conda\r\n\n", "before_files": [{"content": "\"\"\"Utilities for integrating with IPython\n\nThese functions should probably reside in Jupyter and IPython repositories,\nafter which we can import them instead of having our own definitions.\n\"\"\"\n\nimport atexit\nimport os\n\ntry:\n    import queue\nexcept ImportError:\n    # Python 2\n    import Queue as queue\nfrom subprocess import Popen\nimport sys\nfrom threading import Thread\nfrom uuid import uuid4\n\nfrom tornado.gen import TimeoutError\nfrom tornado.ioloop import IOLoop\nfrom threading import Event\n\nfrom IPython import get_ipython\nfrom jupyter_client import BlockingKernelClient, write_connection_file\nfrom jupyter_core.paths import jupyter_runtime_dir\n\n\nOUTPUT_TIMEOUT = 10\n\n\ndef run_cell_remote(ip, kc, cell):\n    \"\"\"Run a cell on a KernelClient\n\n    Any output from the cell will be redisplayed in the local session.\n    \"\"\"\n    msg_id = kc.execute(cell)\n\n    in_kernel = getattr(ip, \"kernel\", False)\n    if in_kernel:\n        socket = ip.display_pub.pub_socket\n        session = ip.display_pub.session\n        parent_header = ip.display_pub.parent_header\n\n    while True:\n        try:\n            msg = kc.get_iopub_msg(timeout=OUTPUT_TIMEOUT)\n        except queue.Empty:\n            raise TimeoutError(\"Timeout waiting for IPython output\")\n\n        if msg[\"parent_header\"].get(\"msg_id\") != msg_id:\n            continue\n        msg_type = msg[\"header\"][\"msg_type\"]\n        content = msg[\"content\"]\n        if msg_type == \"status\":\n            if content[\"execution_state\"] == \"idle\":\n                # idle means output is done\n                break\n        elif msg_type == \"stream\":\n            stream = getattr(sys, content[\"name\"])\n            stream.write(content[\"text\"])\n        elif msg_type in (\"display_data\", \"execute_result\", \"error\"):\n            if in_kernel:\n                session.send(socket, msg_type, content, parent=parent_header)\n            else:\n                if msg_type == \"error\":\n                    print(\"\\n\".join(content[\"traceback\"]), file=sys.stderr)\n                else:\n                    sys.stdout.write(content[\"data\"].get(\"text/plain\", \"\"))\n        else:\n            pass\n\n\ndef register_worker_magic(connection_info, magic_name=\"worker\"):\n    \"\"\"Register a %worker magic, given connection_info.\n\n    Both a line and cell magic are registered,\n    which run the given cell in a remote kernel.\n    \"\"\"\n    ip = get_ipython()\n    info = dict(connection_info)  # copy\n    key = info.pop(\"key\")\n    kc = BlockingKernelClient(**connection_info)\n    kc.session.key = key\n    kc.start_channels()\n\n    def remote(line, cell=None):\n        \"\"\"Run the current cell on a remote IPython kernel\"\"\"\n        if cell is None:\n            # both line and cell magic\n            cell = line\n        run_cell_remote(ip, kc, cell)\n\n    remote.client = kc  # preserve reference on kc, largely for mocking\n    ip.register_magic_function(remote, magic_kind=\"line\", magic_name=magic_name)\n    ip.register_magic_function(remote, magic_kind=\"cell\", magic_name=magic_name)\n\n\ndef remote_magic(line, cell=None):\n    \"\"\"A magic for running code on a specified remote worker\n\n    The connection_info dict of the worker will be looked up\n    as the first positional arg to the magic.\n    The rest of the line (or the entire cell for a %%cell magic)\n    will be passed to the remote kernel.\n\n    Usage:\n\n        info = e.start_ipython(worker)[worker]\n        %remote info print(worker.data)\n    \"\"\"\n    # get connection info from IPython's user namespace\n    ip = get_ipython()\n    split_line = line.split(None, 1)\n    info_name = split_line[0]\n    if info_name not in ip.user_ns:\n        raise NameError(info_name)\n    connection_info = dict(ip.user_ns[info_name])\n\n    if not cell:  # line magic, use the rest of the line\n        if len(split_line) == 1:\n            raise ValueError(\"I need some code to run!\")\n        cell = split_line[1]\n\n    # turn info dict to hashable str for use as lookup key in _clients cache\n    key = \",\".join(map(str, sorted(connection_info.items())))\n    session_key = connection_info.pop(\"key\")\n\n    if key in remote_magic._clients:\n        kc = remote_magic._clients[key]\n    else:\n        kc = BlockingKernelClient(**connection_info)\n        kc.session.key = session_key\n        kc.start_channels()\n        kc.wait_for_ready(timeout=10)\n        remote_magic._clients[key] = kc\n\n    # actually run the code\n    run_cell_remote(ip, kc, cell)\n\n\n# cache clients for re-use in remote magic\nremote_magic._clients = {}\n\n\ndef register_remote_magic(magic_name=\"remote\"):\n    \"\"\"Define the parameterized %remote magic\n\n    See remote_magic above for details.\n    \"\"\"\n    ip = get_ipython()\n    if ip is None:\n        return  # do nothing if IPython's not running\n    ip.register_magic_function(remote_magic, magic_kind=\"line\", magic_name=magic_name)\n    ip.register_magic_function(remote_magic, magic_kind=\"cell\", magic_name=magic_name)\n\n\ndef connect_qtconsole(connection_info, name=None, extra_args=None):\n    \"\"\"Open a QtConsole connected to a worker who has the given future\n\n    - identify worker with who_has\n    - start IPython kernel on the worker\n    - start qtconsole connected to the kernel\n    \"\"\"\n    runtime_dir = jupyter_runtime_dir()\n    if name is None:\n        name = uuid4().hex\n\n    path = os.path.join(runtime_dir, name + \".json\")\n    write_connection_file(path, **connection_info)\n    cmd = [\"jupyter\", \"qtconsole\", \"--existing\", path]\n    if extra_args:\n        cmd.extend(extra_args)\n    Popen(cmd)\n\n    @atexit.register\n    def _cleanup_connection_file():\n        \"\"\"Cleanup our connection file when we exit.\"\"\"\n        try:\n            os.remove(path)\n        except OSError:\n            pass\n\n\ndef start_ipython(ip=None, ns=None, log=None):\n    \"\"\"Start an IPython kernel in a thread\n\n    Parameters\n    ----------\n    ip : str\n        The IP address to listen on (likely the parent object's ip).\n    ns : dict\n        Any names that should be injected into the IPython namespace.\n    log : logger instance\n        Hook up IPython's logging to an existing logger instead of the default.\n    \"\"\"\n    from IPython import get_ipython\n\n    if get_ipython() is not None:\n        raise RuntimeError(\"Cannot start IPython, it's already running.\")\n\n    from zmq.eventloop.ioloop import ZMQIOLoop\n    from ipykernel.kernelapp import IPKernelApp\n\n    # save the global IOLoop instance\n    # since IPython relies on it, but we are going to put it in a thread.\n    save_inst = IOLoop.instance()\n    IOLoop.clear_instance()\n    zmq_loop = ZMQIOLoop()\n    zmq_loop.install()\n\n    # start IPython, disabling its signal handlers that won't work due to running in a thread:\n    app = IPKernelApp.instance(log=log)\n    # Don't connect to the history database\n    app.config.HistoryManager.hist_file = \":memory:\"\n    # listen on all interfaces, so remote clients can connect:\n    if ip:\n        app.ip = ip\n    # disable some signal handling, logging\n\n    def noop():\n        return None\n\n    app.init_signal = noop\n    app.log_connection_info = noop\n\n    # start IPython in a thread\n    # initialization happens in the thread to avoid threading problems\n    # with the sqlite history\n    evt = Event()\n\n    def _start():\n        app.initialize([])\n        app.kernel.pre_handler_hook = noop\n        app.kernel.post_handler_hook = noop\n        app.kernel.start()\n        app.kernel.loop = IOLoop.instance()\n        # save self in the IPython namespace as 'worker'\n        # inject things into the IPython namespace\n        if ns:\n            app.kernel.shell.user_ns.update(ns)\n        evt.set()\n        zmq_loop.start()\n\n    zmq_loop_thread = Thread(target=_start)\n    zmq_loop_thread.daemon = True\n    zmq_loop_thread.start()\n    assert evt.wait(timeout=5), \"IPython didn't start in a reasonable amount of time.\"\n\n    # put the global IOLoop instance back:\n    IOLoop.clear_instance()\n    save_inst.install()\n    return app\n", "path": "distributed/_ipython_utils.py"}], "after_files": [{"content": "\"\"\"Utilities for integrating with IPython\n\nThese functions should probably reside in Jupyter and IPython repositories,\nafter which we can import them instead of having our own definitions.\n\"\"\"\n\nimport atexit\nimport os\n\ntry:\n    import queue\nexcept ImportError:\n    # Python 2\n    import Queue as queue\nfrom subprocess import Popen\nimport sys\nfrom threading import Thread\nfrom uuid import uuid4\n\nfrom tornado.gen import TimeoutError\nfrom tornado.ioloop import IOLoop\nfrom threading import Event\n\nfrom IPython import get_ipython\nfrom jupyter_client import BlockingKernelClient, write_connection_file\nfrom jupyter_core.paths import jupyter_runtime_dir\n\n\nOUTPUT_TIMEOUT = 10\n\n\ndef run_cell_remote(ip, kc, cell):\n    \"\"\"Run a cell on a KernelClient\n\n    Any output from the cell will be redisplayed in the local session.\n    \"\"\"\n    msg_id = kc.execute(cell)\n\n    in_kernel = getattr(ip, \"kernel\", False)\n    if in_kernel:\n        socket = ip.display_pub.pub_socket\n        session = ip.display_pub.session\n        parent_header = ip.display_pub.parent_header\n\n    while True:\n        try:\n            msg = kc.get_iopub_msg(timeout=OUTPUT_TIMEOUT)\n        except queue.Empty:\n            raise TimeoutError(\"Timeout waiting for IPython output\")\n\n        if msg[\"parent_header\"].get(\"msg_id\") != msg_id:\n            continue\n        msg_type = msg[\"header\"][\"msg_type\"]\n        content = msg[\"content\"]\n        if msg_type == \"status\":\n            if content[\"execution_state\"] == \"idle\":\n                # idle means output is done\n                break\n        elif msg_type == \"stream\":\n            stream = getattr(sys, content[\"name\"])\n            stream.write(content[\"text\"])\n        elif msg_type in (\"display_data\", \"execute_result\", \"error\"):\n            if in_kernel:\n                session.send(socket, msg_type, content, parent=parent_header)\n            else:\n                if msg_type == \"error\":\n                    print(\"\\n\".join(content[\"traceback\"]), file=sys.stderr)\n                else:\n                    sys.stdout.write(content[\"data\"].get(\"text/plain\", \"\"))\n        else:\n            pass\n\n\ndef register_worker_magic(connection_info, magic_name=\"worker\"):\n    \"\"\"Register a %worker magic, given connection_info.\n\n    Both a line and cell magic are registered,\n    which run the given cell in a remote kernel.\n    \"\"\"\n    ip = get_ipython()\n    info = dict(connection_info)  # copy\n    key = info.pop(\"key\")\n    kc = BlockingKernelClient(**connection_info)\n    kc.session.key = key\n    kc.start_channels()\n\n    def remote(line, cell=None):\n        \"\"\"Run the current cell on a remote IPython kernel\"\"\"\n        if cell is None:\n            # both line and cell magic\n            cell = line\n        run_cell_remote(ip, kc, cell)\n\n    remote.client = kc  # preserve reference on kc, largely for mocking\n    ip.register_magic_function(remote, magic_kind=\"line\", magic_name=magic_name)\n    ip.register_magic_function(remote, magic_kind=\"cell\", magic_name=magic_name)\n\n\ndef remote_magic(line, cell=None):\n    \"\"\"A magic for running code on a specified remote worker\n\n    The connection_info dict of the worker will be looked up\n    as the first positional arg to the magic.\n    The rest of the line (or the entire cell for a %%cell magic)\n    will be passed to the remote kernel.\n\n    Usage:\n\n        info = e.start_ipython(worker)[worker]\n        %remote info print(worker.data)\n    \"\"\"\n    # get connection info from IPython's user namespace\n    ip = get_ipython()\n    split_line = line.split(None, 1)\n    info_name = split_line[0]\n    if info_name not in ip.user_ns:\n        raise NameError(info_name)\n    connection_info = dict(ip.user_ns[info_name])\n\n    if not cell:  # line magic, use the rest of the line\n        if len(split_line) == 1:\n            raise ValueError(\"I need some code to run!\")\n        cell = split_line[1]\n\n    # turn info dict to hashable str for use as lookup key in _clients cache\n    key = \",\".join(map(str, sorted(connection_info.items())))\n    session_key = connection_info.pop(\"key\")\n\n    if key in remote_magic._clients:\n        kc = remote_magic._clients[key]\n    else:\n        kc = BlockingKernelClient(**connection_info)\n        kc.session.key = session_key\n        kc.start_channels()\n        kc.wait_for_ready(timeout=10)\n        remote_magic._clients[key] = kc\n\n    # actually run the code\n    run_cell_remote(ip, kc, cell)\n\n\n# cache clients for re-use in remote magic\nremote_magic._clients = {}\n\n\ndef register_remote_magic(magic_name=\"remote\"):\n    \"\"\"Define the parameterized %remote magic\n\n    See remote_magic above for details.\n    \"\"\"\n    ip = get_ipython()\n    if ip is None:\n        return  # do nothing if IPython's not running\n    ip.register_magic_function(remote_magic, magic_kind=\"line\", magic_name=magic_name)\n    ip.register_magic_function(remote_magic, magic_kind=\"cell\", magic_name=magic_name)\n\n\ndef connect_qtconsole(connection_info, name=None, extra_args=None):\n    \"\"\"Open a QtConsole connected to a worker who has the given future\n\n    - identify worker with who_has\n    - start IPython kernel on the worker\n    - start qtconsole connected to the kernel\n    \"\"\"\n    runtime_dir = jupyter_runtime_dir()\n    if name is None:\n        name = uuid4().hex\n\n    path = os.path.join(runtime_dir, name + \".json\")\n    write_connection_file(path, **connection_info)\n    cmd = [\"jupyter\", \"qtconsole\", \"--existing\", path]\n    if extra_args:\n        cmd.extend(extra_args)\n    Popen(cmd)\n\n    @atexit.register\n    def _cleanup_connection_file():\n        \"\"\"Cleanup our connection file when we exit.\"\"\"\n        try:\n            os.remove(path)\n        except OSError:\n            pass\n\n\ndef start_ipython(ip=None, ns=None, log=None):\n    \"\"\"Start an IPython kernel in a thread\n\n    Parameters\n    ----------\n    ip : str\n        The IP address to listen on (likely the parent object's ip).\n    ns : dict\n        Any names that should be injected into the IPython namespace.\n    log : logger instance\n        Hook up IPython's logging to an existing logger instead of the default.\n    \"\"\"\n    from IPython import get_ipython\n\n    if get_ipython() is not None:\n        raise RuntimeError(\"Cannot start IPython, it's already running.\")\n\n    from ipykernel.kernelapp import IPKernelApp\n\n    # start IPython, disabling its signal handlers that won't work due to running in a thread:\n    app = IPKernelApp.instance(log=log)\n    # Don't connect to the history database\n    app.config.HistoryManager.hist_file = \":memory:\"\n    # listen on all interfaces, so remote clients can connect:\n    if ip:\n        app.ip = ip\n    # disable some signal handling, logging\n\n    def noop():\n        return None\n\n    app.init_signal = noop\n    app.log_connection_info = noop\n\n    # start IPython in a thread\n    # initialization happens in the thread to avoid threading problems\n    # with the sqlite history\n    evt = Event()\n\n    def _start():\n        app.initialize([])\n        app.kernel.pre_handler_hook = noop\n        app.kernel.post_handler_hook = noop\n        app.kernel.start()\n        # save self in the IPython namespace as 'worker'\n        # inject things into the IPython namespace\n        if ns:\n            app.kernel.shell.user_ns.update(ns)\n        evt.set()\n        # start the app's IOLoop in its thread\n        IOLoop.current().start()\n\n    zmq_loop_thread = Thread(target=_start)\n    zmq_loop_thread.daemon = True\n    zmq_loop_thread.start()\n    assert evt.wait(timeout=5), \"IPython didn't start in a reasonable amount of time.\"\n\n    return app\n", "path": "distributed/_ipython_utils.py"}]}