##// END OF EJS Templates
improve default ipc file locations...
MinRK -
Show More
@@ -1,351 +1,371
1 """An Application for launching a kernel
1 """An Application for launching a kernel
2
2
3 Authors
3 Authors
4 -------
4 -------
5 * MinRK
5 * MinRK
6 """
6 """
7 #-----------------------------------------------------------------------------
7 #-----------------------------------------------------------------------------
8 # Copyright (C) 2011 The IPython Development Team
8 # Copyright (C) 2011 The IPython Development Team
9 #
9 #
10 # Distributed under the terms of the BSD License. The full license is in
10 # Distributed under the terms of the BSD License. The full license is in
11 # the file COPYING.txt, distributed as part of this software.
11 # the file COPYING.txt, distributed as part of this software.
12 #-----------------------------------------------------------------------------
12 #-----------------------------------------------------------------------------
13
13
14 #-----------------------------------------------------------------------------
14 #-----------------------------------------------------------------------------
15 # Imports
15 # Imports
16 #-----------------------------------------------------------------------------
16 #-----------------------------------------------------------------------------
17
17
18 # Standard library imports
18 # Standard library imports
19 import atexit
19 import atexit
20 import json
20 import json
21 import os
21 import os
22 import sys
22 import sys
23 import signal
23 import signal
24
24
25 # System library imports
25 # System library imports
26 import zmq
26 import zmq
27 from zmq.eventloop import ioloop
27 from zmq.eventloop import ioloop
28
28
29 # IPython imports
29 # IPython imports
30 from IPython.core.ultratb import FormattedTB
30 from IPython.core.ultratb import FormattedTB
31 from IPython.core.application import (
31 from IPython.core.application import (
32 BaseIPythonApplication, base_flags, base_aliases, catch_config_error
32 BaseIPythonApplication, base_flags, base_aliases, catch_config_error
33 )
33 )
34 from IPython.utils import io
34 from IPython.utils import io
35 from IPython.utils.localinterfaces import LOCALHOST
35 from IPython.utils.localinterfaces import LOCALHOST
36 from IPython.utils.path import filefind
36 from IPython.utils.path import filefind
37 from IPython.utils.py3compat import str_to_bytes
37 from IPython.utils.py3compat import str_to_bytes
38 from IPython.utils.traitlets import (
38 from IPython.utils.traitlets import (
39 Any, Instance, Dict, Unicode, Integer, Bool, CaselessStrEnum,
39 Any, Instance, Dict, Unicode, Integer, Bool, CaselessStrEnum,
40 DottedObjectName,
40 DottedObjectName,
41 )
41 )
42 from IPython.utils.importstring import import_item
42 from IPython.utils.importstring import import_item
43 # local imports
43 # local imports
44 from IPython.zmq.entry_point import write_connection_file
44 from IPython.zmq.entry_point import write_connection_file
45 from IPython.zmq.heartbeat import Heartbeat
45 from IPython.zmq.heartbeat import Heartbeat
46 from IPython.zmq.parentpoller import ParentPollerUnix, ParentPollerWindows
46 from IPython.zmq.parentpoller import ParentPollerUnix, ParentPollerWindows
47 from IPython.zmq.session import (
47 from IPython.zmq.session import (
48 Session, session_flags, session_aliases, default_secure,
48 Session, session_flags, session_aliases, default_secure,
49 )
49 )
50
50
51
51
52 #-----------------------------------------------------------------------------
52 #-----------------------------------------------------------------------------
53 # Flags and Aliases
53 # Flags and Aliases
54 #-----------------------------------------------------------------------------
54 #-----------------------------------------------------------------------------
55
55
56 kernel_aliases = dict(base_aliases)
56 kernel_aliases = dict(base_aliases)
57 kernel_aliases.update({
57 kernel_aliases.update({
58 'ip' : 'KernelApp.ip',
58 'ip' : 'KernelApp.ip',
59 'hb' : 'KernelApp.hb_port',
59 'hb' : 'KernelApp.hb_port',
60 'shell' : 'KernelApp.shell_port',
60 'shell' : 'KernelApp.shell_port',
61 'iopub' : 'KernelApp.iopub_port',
61 'iopub' : 'KernelApp.iopub_port',
62 'stdin' : 'KernelApp.stdin_port',
62 'stdin' : 'KernelApp.stdin_port',
63 'f' : 'KernelApp.connection_file',
63 'f' : 'KernelApp.connection_file',
64 'parent': 'KernelApp.parent',
64 'parent': 'KernelApp.parent',
65 'transport': 'KernelApp.transport',
65 })
66 })
66 if sys.platform.startswith('win'):
67 if sys.platform.startswith('win'):
67 kernel_aliases['interrupt'] = 'KernelApp.interrupt'
68 kernel_aliases['interrupt'] = 'KernelApp.interrupt'
68
69
69 kernel_flags = dict(base_flags)
70 kernel_flags = dict(base_flags)
70 kernel_flags.update({
71 kernel_flags.update({
71 'no-stdout' : (
72 'no-stdout' : (
72 {'KernelApp' : {'no_stdout' : True}},
73 {'KernelApp' : {'no_stdout' : True}},
73 "redirect stdout to the null device"),
74 "redirect stdout to the null device"),
74 'no-stderr' : (
75 'no-stderr' : (
75 {'KernelApp' : {'no_stderr' : True}},
76 {'KernelApp' : {'no_stderr' : True}},
76 "redirect stderr to the null device"),
77 "redirect stderr to the null device"),
77 })
78 })
78
79
79 # inherit flags&aliases for Sessions
80 # inherit flags&aliases for Sessions
80 kernel_aliases.update(session_aliases)
81 kernel_aliases.update(session_aliases)
81 kernel_flags.update(session_flags)
82 kernel_flags.update(session_flags)
82
83
83
84
84
85
85 #-----------------------------------------------------------------------------
86 #-----------------------------------------------------------------------------
86 # Application class for starting a Kernel
87 # Application class for starting a Kernel
87 #-----------------------------------------------------------------------------
88 #-----------------------------------------------------------------------------
88
89
89 class KernelApp(BaseIPythonApplication):
90 class KernelApp(BaseIPythonApplication):
90 name='ipkernel'
91 name='ipkernel'
91 aliases = Dict(kernel_aliases)
92 aliases = Dict(kernel_aliases)
92 flags = Dict(kernel_flags)
93 flags = Dict(kernel_flags)
93 classes = [Session]
94 classes = [Session]
94 # the kernel class, as an importstring
95 # the kernel class, as an importstring
95 kernel_class = DottedObjectName('IPython.zmq.ipkernel.Kernel')
96 kernel_class = DottedObjectName('IPython.zmq.ipkernel.Kernel')
96 kernel = Any()
97 kernel = Any()
97 poller = Any() # don't restrict this even though current pollers are all Threads
98 poller = Any() # don't restrict this even though current pollers are all Threads
98 heartbeat = Instance(Heartbeat)
99 heartbeat = Instance(Heartbeat)
99 session = Instance('IPython.zmq.session.Session')
100 session = Instance('IPython.zmq.session.Session')
100 ports = Dict()
101 ports = Dict()
101 _full_connection_file = Unicode()
102
102
103 # inherit config file name from parent:
103 # inherit config file name from parent:
104 parent_appname = Unicode(config=True)
104 parent_appname = Unicode(config=True)
105 def _parent_appname_changed(self, name, old, new):
105 def _parent_appname_changed(self, name, old, new):
106 if self.config_file_specified:
106 if self.config_file_specified:
107 # it was manually specified, ignore
107 # it was manually specified, ignore
108 return
108 return
109 self.config_file_name = new.replace('-','_') + u'_config.py'
109 self.config_file_name = new.replace('-','_') + u'_config.py'
110 # don't let this count as specifying the config file
110 # don't let this count as specifying the config file
111 self.config_file_specified = False
111 self.config_file_specified = False
112
112
113 # connection info:
113 # connection info:
114 transport = CaselessStrEnum(['tcp', 'ipc'], default_value='tcp', config=True)
114 transport = CaselessStrEnum(['tcp', 'ipc'], default_value='tcp', config=True)
115 ip = Unicode(LOCALHOST, config=True,
115 ip = Unicode(config=True,
116 help="Set the IP or interface on which the kernel will listen.")
116 help="Set the IP or interface on which the kernel will listen.")
117 def _ip_default(self):
118 if self.transport == 'ipc':
119 if self.connection_file:
120 return os.path.splitext(self.abs_connection_file)[0] + '-ipc'
121 else:
122 return 'kernel-ipc'
123 else:
124 return LOCALHOST
117 hb_port = Integer(0, config=True, help="set the heartbeat port [default: random]")
125 hb_port = Integer(0, config=True, help="set the heartbeat port [default: random]")
118 shell_port = Integer(0, config=True, help="set the shell (ROUTER) port [default: random]")
126 shell_port = Integer(0, config=True, help="set the shell (ROUTER) port [default: random]")
119 iopub_port = Integer(0, config=True, help="set the iopub (PUB) port [default: random]")
127 iopub_port = Integer(0, config=True, help="set the iopub (PUB) port [default: random]")
120 stdin_port = Integer(0, config=True, help="set the stdin (DEALER) port [default: random]")
128 stdin_port = Integer(0, config=True, help="set the stdin (DEALER) port [default: random]")
121 connection_file = Unicode('', config=True,
129 connection_file = Unicode('', config=True,
122 help="""JSON file in which to store connection info [default: kernel-<pid>.json]
130 help="""JSON file in which to store connection info [default: kernel-<pid>.json]
123
131
124 This file will contain the IP, ports, and authentication key needed to connect
132 This file will contain the IP, ports, and authentication key needed to connect
125 clients to this kernel. By default, this file will be created in the security-dir
133 clients to this kernel. By default, this file will be created in the security dir
126 of the current profile, but can be specified by absolute path.
134 of the current profile, but can be specified by absolute path.
127 """)
135 """)
136 @property
137 def abs_connection_file(self):
138 if os.path.basename(self.connection_file) == self.connection_file:
139 return os.path.join(self.profile_dir.security_dir, self.connection_file)
140 else:
141 return self.connection_file
142
128
143
129 # streams, etc.
144 # streams, etc.
130 no_stdout = Bool(False, config=True, help="redirect stdout to the null device")
145 no_stdout = Bool(False, config=True, help="redirect stdout to the null device")
131 no_stderr = Bool(False, config=True, help="redirect stderr to the null device")
146 no_stderr = Bool(False, config=True, help="redirect stderr to the null device")
132 outstream_class = DottedObjectName('IPython.zmq.iostream.OutStream',
147 outstream_class = DottedObjectName('IPython.zmq.iostream.OutStream',
133 config=True, help="The importstring for the OutStream factory")
148 config=True, help="The importstring for the OutStream factory")
134 displayhook_class = DottedObjectName('IPython.zmq.displayhook.ZMQDisplayHook',
149 displayhook_class = DottedObjectName('IPython.zmq.displayhook.ZMQDisplayHook',
135 config=True, help="The importstring for the DisplayHook factory")
150 config=True, help="The importstring for the DisplayHook factory")
136
151
137 # polling
152 # polling
138 parent = Integer(0, config=True,
153 parent = Integer(0, config=True,
139 help="""kill this process if its parent dies. On Windows, the argument
154 help="""kill this process if its parent dies. On Windows, the argument
140 specifies the HANDLE of the parent process, otherwise it is simply boolean.
155 specifies the HANDLE of the parent process, otherwise it is simply boolean.
141 """)
156 """)
142 interrupt = Integer(0, config=True,
157 interrupt = Integer(0, config=True,
143 help="""ONLY USED ON WINDOWS
158 help="""ONLY USED ON WINDOWS
144 Interrupt this process when the parent is signalled.
159 Interrupt this process when the parent is signaled.
145 """)
160 """)
146
161
147 def init_crash_handler(self):
162 def init_crash_handler(self):
148 # Install minimal exception handling
163 # Install minimal exception handling
149 sys.excepthook = FormattedTB(mode='Verbose', color_scheme='NoColor',
164 sys.excepthook = FormattedTB(mode='Verbose', color_scheme='NoColor',
150 ostream=sys.__stdout__)
165 ostream=sys.__stdout__)
151
166
152 def init_poller(self):
167 def init_poller(self):
153 if sys.platform == 'win32':
168 if sys.platform == 'win32':
154 if self.interrupt or self.parent:
169 if self.interrupt or self.parent:
155 self.poller = ParentPollerWindows(self.interrupt, self.parent)
170 self.poller = ParentPollerWindows(self.interrupt, self.parent)
156 elif self.parent:
171 elif self.parent:
157 self.poller = ParentPollerUnix()
172 self.poller = ParentPollerUnix()
158
173
159 def _bind_socket(self, s, port):
174 def _bind_socket(self, s, port):
160 iface = '%s://%s' % (self.transport, self.ip)
175 iface = '%s://%s' % (self.transport, self.ip)
161 if port <= 0 and self.transport == 'tcp':
176 if self.transport == 'tcp':
177 if port <= 0:
162 port = s.bind_to_random_port(iface)
178 port = s.bind_to_random_port(iface)
163 else:
179 else:
164 c = ':' if self.transport == 'tcp' else '-'
180 s.bind("tcp://%s:%i" % (self.ip, port))
165 s.bind(iface + c + str(port))
181 elif self.transport == 'ipc':
182 if port <= 0:
183 for port in range(1,1024):
184 path = "%s-%i" % (self.ip, port)
185 if not os.path.exists(path):
186 break
187 else:
188 path = "%s-%i" % (self.ip, port)
189 s.bind("ipc://%s" % path)
166 return port
190 return port
167
191
168 def load_connection_file(self):
192 def load_connection_file(self):
169 """load ip/port/hmac config from JSON connection file"""
193 """load ip/port/hmac config from JSON connection file"""
170 try:
194 try:
171 fname = filefind(self.connection_file, ['.', self.profile_dir.security_dir])
195 fname = filefind(self.connection_file, ['.', self.profile_dir.security_dir])
172 except IOError:
196 except IOError:
173 self.log.debug("Connection file not found: %s", self.connection_file)
197 self.log.debug("Connection file not found: %s", self.connection_file)
174 # This means I own it, so I will clean it up:
198 # This means I own it, so I will clean it up:
175 atexit.register(self.cleanup_connection_file)
199 atexit.register(self.cleanup_connection_file)
176 return
200 return
177 self.log.debug(u"Loading connection file %s", fname)
201 self.log.debug(u"Loading connection file %s", fname)
178 with open(fname) as f:
202 with open(fname) as f:
179 s = f.read()
203 s = f.read()
180 cfg = json.loads(s)
204 cfg = json.loads(s)
181 self.transport = cfg.get('transport', self.transport)
205 self.transport = cfg.get('transport', self.transport)
182 if self.ip == LOCALHOST and 'ip' in cfg:
206 if self.ip == self._ip_default() and 'ip' in cfg:
183 # not overridden by config or cl_args
207 # not overridden by config or cl_args
184 self.ip = cfg['ip']
208 self.ip = cfg['ip']
185 for channel in ('hb', 'shell', 'iopub', 'stdin'):
209 for channel in ('hb', 'shell', 'iopub', 'stdin'):
186 name = channel + '_port'
210 name = channel + '_port'
187 if getattr(self, name) == 0 and name in cfg:
211 if getattr(self, name) == 0 and name in cfg:
188 # not overridden by config or cl_args
212 # not overridden by config or cl_args
189 setattr(self, name, cfg[name])
213 setattr(self, name, cfg[name])
190 if 'key' in cfg:
214 if 'key' in cfg:
191 self.config.Session.key = str_to_bytes(cfg['key'])
215 self.config.Session.key = str_to_bytes(cfg['key'])
192
216
193 def write_connection_file(self):
217 def write_connection_file(self):
194 """write connection info to JSON file"""
218 """write connection info to JSON file"""
195 if os.path.basename(self.connection_file) == self.connection_file:
219 cf = self.abs_connection_file
196 cf = os.path.join(self.profile_dir.security_dir, self.connection_file)
220 self.log.debug("Writing connection file: %s", cf)
197 else:
198 cf = self.connection_file
199 write_connection_file(cf, ip=self.ip, key=self.session.key, transport=self.transport,
221 write_connection_file(cf, ip=self.ip, key=self.session.key, transport=self.transport,
200 shell_port=self.shell_port, stdin_port=self.stdin_port, hb_port=self.hb_port,
222 shell_port=self.shell_port, stdin_port=self.stdin_port, hb_port=self.hb_port,
201 iopub_port=self.iopub_port)
223 iopub_port=self.iopub_port)
202
224
203 self._full_connection_file = cf
204
205 def cleanup_connection_file(self):
225 def cleanup_connection_file(self):
206 cf = self._full_connection_file
226 cf = self.abs_connection_file
207 self.log.debug("cleaning up connection file: %r", cf)
227 self.log.debug("Cleaning up connection file: %s", cf)
208 try:
228 try:
209 os.remove(cf)
229 os.remove(cf)
210 except (IOError, OSError):
230 except (IOError, OSError):
211 pass
231 pass
212
232
213 self.cleanup_ipc_files()
233 self.cleanup_ipc_files()
214
234
215 def cleanup_ipc_files(self):
235 def cleanup_ipc_files(self):
216 """cleanup ipc files if we wrote them"""
236 """cleanup ipc files if we wrote them"""
217 if self.transport != 'ipc':
237 if self.transport != 'ipc':
218 return
238 return
219 for port in (self.shell_port, self.iopub_port, self.stdin_port, self.hb_port):
239 for port in (self.shell_port, self.iopub_port, self.stdin_port, self.hb_port):
220 ipcfile = "%s-%i" % (self.ip, port)
240 ipcfile = "%s-%i" % (self.ip, port)
221 try:
241 try:
222 os.remove(ipcfile)
242 os.remove(ipcfile)
223 except (IOError, OSError):
243 except (IOError, OSError):
224 pass
244 pass
225
245
226 def init_connection_file(self):
246 def init_connection_file(self):
227 if not self.connection_file:
247 if not self.connection_file:
228 self.connection_file = "kernel-%s.json"%os.getpid()
248 self.connection_file = "kernel-%s.json"%os.getpid()
229 try:
249 try:
230 self.load_connection_file()
250 self.load_connection_file()
231 except Exception:
251 except Exception:
232 self.log.error("Failed to load connection file: %r", self.connection_file, exc_info=True)
252 self.log.error("Failed to load connection file: %r", self.connection_file, exc_info=True)
233 self.exit(1)
253 self.exit(1)
234
254
235 def init_sockets(self):
255 def init_sockets(self):
236 # Create a context, a session, and the kernel sockets.
256 # Create a context, a session, and the kernel sockets.
237 self.log.info("Starting the kernel at pid: %i", os.getpid())
257 self.log.info("Starting the kernel at pid: %i", os.getpid())
238 context = zmq.Context.instance()
258 context = zmq.Context.instance()
239 # Uncomment this to try closing the context.
259 # Uncomment this to try closing the context.
240 # atexit.register(context.term)
260 # atexit.register(context.term)
241
261
242 self.shell_socket = context.socket(zmq.ROUTER)
262 self.shell_socket = context.socket(zmq.ROUTER)
243 self.shell_port = self._bind_socket(self.shell_socket, self.shell_port)
263 self.shell_port = self._bind_socket(self.shell_socket, self.shell_port)
244 self.log.debug("shell ROUTER Channel on port: %i"%self.shell_port)
264 self.log.debug("shell ROUTER Channel on port: %i"%self.shell_port)
245
265
246 self.iopub_socket = context.socket(zmq.PUB)
266 self.iopub_socket = context.socket(zmq.PUB)
247 self.iopub_port = self._bind_socket(self.iopub_socket, self.iopub_port)
267 self.iopub_port = self._bind_socket(self.iopub_socket, self.iopub_port)
248 self.log.debug("iopub PUB Channel on port: %i"%self.iopub_port)
268 self.log.debug("iopub PUB Channel on port: %i"%self.iopub_port)
249
269
250 self.stdin_socket = context.socket(zmq.ROUTER)
270 self.stdin_socket = context.socket(zmq.ROUTER)
251 self.stdin_port = self._bind_socket(self.stdin_socket, self.stdin_port)
271 self.stdin_port = self._bind_socket(self.stdin_socket, self.stdin_port)
252 self.log.debug("stdin ROUTER Channel on port: %i"%self.stdin_port)
272 self.log.debug("stdin ROUTER Channel on port: %i"%self.stdin_port)
253
273
254 def init_heartbeat(self):
274 def init_heartbeat(self):
255 """start the heart beating"""
275 """start the heart beating"""
256 # heartbeat doesn't share context, because it mustn't be blocked
276 # heartbeat doesn't share context, because it mustn't be blocked
257 # by the GIL, which is accessed by libzmq when freeing zero-copy messages
277 # by the GIL, which is accessed by libzmq when freeing zero-copy messages
258 hb_ctx = zmq.Context()
278 hb_ctx = zmq.Context()
259 self.heartbeat = Heartbeat(hb_ctx, (self.transport, self.ip, self.hb_port))
279 self.heartbeat = Heartbeat(hb_ctx, (self.transport, self.ip, self.hb_port))
260 self.hb_port = self.heartbeat.port
280 self.hb_port = self.heartbeat.port
261 self.log.debug("Heartbeat REP Channel on port: %i"%self.hb_port)
281 self.log.debug("Heartbeat REP Channel on port: %i"%self.hb_port)
262 self.heartbeat.start()
282 self.heartbeat.start()
263
283
264 # Helper to make it easier to connect to an existing kernel.
284 # Helper to make it easier to connect to an existing kernel.
265 # set log-level to critical, to make sure it is output
285 # set log-level to critical, to make sure it is output
266 self.log.critical("To connect another client to this kernel, use:")
286 self.log.critical("To connect another client to this kernel, use:")
267
287
268 def log_connection_info(self):
288 def log_connection_info(self):
269 """display connection info, and store ports"""
289 """display connection info, and store ports"""
270 basename = os.path.basename(self.connection_file)
290 basename = os.path.basename(self.connection_file)
271 if basename == self.connection_file or \
291 if basename == self.connection_file or \
272 os.path.dirname(self.connection_file) == self.profile_dir.security_dir:
292 os.path.dirname(self.connection_file) == self.profile_dir.security_dir:
273 # use shortname
293 # use shortname
274 tail = basename
294 tail = basename
275 if self.profile != 'default':
295 if self.profile != 'default':
276 tail += " --profile %s" % self.profile
296 tail += " --profile %s" % self.profile
277 else:
297 else:
278 tail = self.connection_file
298 tail = self.connection_file
279 self.log.critical("--existing %s", tail)
299 self.log.critical("--existing %s", tail)
280
300
281
301
282 self.ports = dict(shell=self.shell_port, iopub=self.iopub_port,
302 self.ports = dict(shell=self.shell_port, iopub=self.iopub_port,
283 stdin=self.stdin_port, hb=self.hb_port)
303 stdin=self.stdin_port, hb=self.hb_port)
284
304
285 def init_session(self):
305 def init_session(self):
286 """create our session object"""
306 """create our session object"""
287 default_secure(self.config)
307 default_secure(self.config)
288 self.session = Session(config=self.config, username=u'kernel')
308 self.session = Session(config=self.config, username=u'kernel')
289
309
290 def init_blackhole(self):
310 def init_blackhole(self):
291 """redirects stdout/stderr to devnull if necessary"""
311 """redirects stdout/stderr to devnull if necessary"""
292 if self.no_stdout or self.no_stderr:
312 if self.no_stdout or self.no_stderr:
293 blackhole = open(os.devnull, 'w')
313 blackhole = open(os.devnull, 'w')
294 if self.no_stdout:
314 if self.no_stdout:
295 sys.stdout = sys.__stdout__ = blackhole
315 sys.stdout = sys.__stdout__ = blackhole
296 if self.no_stderr:
316 if self.no_stderr:
297 sys.stderr = sys.__stderr__ = blackhole
317 sys.stderr = sys.__stderr__ = blackhole
298
318
299 def init_io(self):
319 def init_io(self):
300 """Redirect input streams and set a display hook."""
320 """Redirect input streams and set a display hook."""
301 if self.outstream_class:
321 if self.outstream_class:
302 outstream_factory = import_item(str(self.outstream_class))
322 outstream_factory = import_item(str(self.outstream_class))
303 sys.stdout = outstream_factory(self.session, self.iopub_socket, u'stdout')
323 sys.stdout = outstream_factory(self.session, self.iopub_socket, u'stdout')
304 sys.stderr = outstream_factory(self.session, self.iopub_socket, u'stderr')
324 sys.stderr = outstream_factory(self.session, self.iopub_socket, u'stderr')
305 if self.displayhook_class:
325 if self.displayhook_class:
306 displayhook_factory = import_item(str(self.displayhook_class))
326 displayhook_factory = import_item(str(self.displayhook_class))
307 sys.displayhook = displayhook_factory(self.session, self.iopub_socket)
327 sys.displayhook = displayhook_factory(self.session, self.iopub_socket)
308
328
309 def init_signal(self):
329 def init_signal(self):
310 signal.signal(signal.SIGINT, signal.SIG_IGN)
330 signal.signal(signal.SIGINT, signal.SIG_IGN)
311
331
312 def init_kernel(self):
332 def init_kernel(self):
313 """Create the Kernel object itself"""
333 """Create the Kernel object itself"""
314 kernel_factory = import_item(str(self.kernel_class))
334 kernel_factory = import_item(str(self.kernel_class))
315 self.kernel = kernel_factory(config=self.config, session=self.session,
335 self.kernel = kernel_factory(config=self.config, session=self.session,
316 shell_socket=self.shell_socket,
336 shell_socket=self.shell_socket,
317 iopub_socket=self.iopub_socket,
337 iopub_socket=self.iopub_socket,
318 stdin_socket=self.stdin_socket,
338 stdin_socket=self.stdin_socket,
319 log=self.log
339 log=self.log
320 )
340 )
321 self.kernel.record_ports(self.ports)
341 self.kernel.record_ports(self.ports)
322
342
323 @catch_config_error
343 @catch_config_error
324 def initialize(self, argv=None):
344 def initialize(self, argv=None):
325 super(KernelApp, self).initialize(argv)
345 super(KernelApp, self).initialize(argv)
326 self.init_blackhole()
346 self.init_blackhole()
327 self.init_connection_file()
347 self.init_connection_file()
328 self.init_session()
348 self.init_session()
329 self.init_poller()
349 self.init_poller()
330 self.init_sockets()
350 self.init_sockets()
331 self.init_heartbeat()
351 self.init_heartbeat()
332 # writing/displaying connection info must be *after* init_sockets/heartbeat
352 # writing/displaying connection info must be *after* init_sockets/heartbeat
333 self.log_connection_info()
353 self.log_connection_info()
334 self.write_connection_file()
354 self.write_connection_file()
335 self.init_io()
355 self.init_io()
336 self.init_signal()
356 self.init_signal()
337 self.init_kernel()
357 self.init_kernel()
338 # flush stdout/stderr, so that anything written to these streams during
358 # flush stdout/stderr, so that anything written to these streams during
339 # initialization do not get associated with the first execution request
359 # initialization do not get associated with the first execution request
340 sys.stdout.flush()
360 sys.stdout.flush()
341 sys.stderr.flush()
361 sys.stderr.flush()
342
362
343 def start(self):
363 def start(self):
344 if self.poller is not None:
364 if self.poller is not None:
345 self.poller.start()
365 self.poller.start()
346 self.kernel.start()
366 self.kernel.start()
347 try:
367 try:
348 ioloop.IOLoop.instance().start()
368 ioloop.IOLoop.instance().start()
349 except KeyboardInterrupt:
369 except KeyboardInterrupt:
350 pass
370 pass
351
371
@@ -1,1081 +1,1094
1 """Base classes to manage the interaction with a running kernel.
1 """Base classes to manage the interaction with a running kernel.
2
2
3 TODO
3 TODO
4 * Create logger to handle debugging and console messages.
4 * Create logger to handle debugging and console messages.
5 """
5 """
6
6
7 #-----------------------------------------------------------------------------
7 #-----------------------------------------------------------------------------
8 # Copyright (C) 2008-2011 The IPython Development Team
8 # Copyright (C) 2008-2011 The IPython Development Team
9 #
9 #
10 # Distributed under the terms of the BSD License. The full license is in
10 # Distributed under the terms of the BSD License. The full license is in
11 # the file COPYING, distributed as part of this software.
11 # the file COPYING, distributed as part of this software.
12 #-----------------------------------------------------------------------------
12 #-----------------------------------------------------------------------------
13
13
14 #-----------------------------------------------------------------------------
14 #-----------------------------------------------------------------------------
15 # Imports
15 # Imports
16 #-----------------------------------------------------------------------------
16 #-----------------------------------------------------------------------------
17
17
18 # Standard library imports.
18 # Standard library imports.
19 import atexit
19 import atexit
20 import errno
20 import errno
21 import json
21 import json
22 from subprocess import Popen
22 from subprocess import Popen
23 import os
23 import os
24 import signal
24 import signal
25 import sys
25 import sys
26 from threading import Thread
26 from threading import Thread
27 import time
27 import time
28
28
29 # System library imports.
29 # System library imports.
30 import zmq
30 import zmq
31 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
31 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
32 # during garbage collection of threads at exit:
32 # during garbage collection of threads at exit:
33 from zmq import ZMQError
33 from zmq import ZMQError
34 from zmq.eventloop import ioloop, zmqstream
34 from zmq.eventloop import ioloop, zmqstream
35
35
36 # Local imports.
36 # Local imports.
37 from IPython.config.configurable import Configurable
37 from IPython.config.configurable import Configurable
38 from IPython.utils.localinterfaces import LOCALHOST, LOCAL_IPS
38 from IPython.utils.localinterfaces import LOCALHOST, LOCAL_IPS
39 from IPython.utils.traitlets import (
39 from IPython.utils.traitlets import (
40 Any, Instance, Type, Unicode, Integer, Bool, CaselessStrEnum
40 Any, Instance, Type, Unicode, Integer, Bool, CaselessStrEnum
41 )
41 )
42 from IPython.utils.py3compat import str_to_bytes
42 from IPython.utils.py3compat import str_to_bytes
43 from IPython.zmq.entry_point import write_connection_file
43 from IPython.zmq.entry_point import write_connection_file
44 from session import Session
44 from session import Session
45 from IPython.zmq.kernelmanagerabc import (
45 from IPython.zmq.kernelmanagerabc import (
46 ShellChannelABC, IOPubChannelABC,
46 ShellChannelABC, IOPubChannelABC,
47 HBChannelABC, StdInChannelABC,
47 HBChannelABC, StdInChannelABC,
48 KernelManagerABC
48 KernelManagerABC
49 )
49 )
50
50
51
51
52 #-----------------------------------------------------------------------------
52 #-----------------------------------------------------------------------------
53 # Constants and exceptions
53 # Constants and exceptions
54 #-----------------------------------------------------------------------------
54 #-----------------------------------------------------------------------------
55
55
56 class InvalidPortNumber(Exception):
56 class InvalidPortNumber(Exception):
57 pass
57 pass
58
58
59 #-----------------------------------------------------------------------------
59 #-----------------------------------------------------------------------------
60 # Utility functions
60 # Utility functions
61 #-----------------------------------------------------------------------------
61 #-----------------------------------------------------------------------------
62
62
63 # some utilities to validate message structure, these might get moved elsewhere
63 # some utilities to validate message structure, these might get moved elsewhere
64 # if they prove to have more generic utility
64 # if they prove to have more generic utility
65
65
66 def validate_string_list(lst):
66 def validate_string_list(lst):
67 """Validate that the input is a list of strings.
67 """Validate that the input is a list of strings.
68
68
69 Raises ValueError if not."""
69 Raises ValueError if not."""
70 if not isinstance(lst, list):
70 if not isinstance(lst, list):
71 raise ValueError('input %r must be a list' % lst)
71 raise ValueError('input %r must be a list' % lst)
72 for x in lst:
72 for x in lst:
73 if not isinstance(x, basestring):
73 if not isinstance(x, basestring):
74 raise ValueError('element %r in list must be a string' % x)
74 raise ValueError('element %r in list must be a string' % x)
75
75
76
76
77 def validate_string_dict(dct):
77 def validate_string_dict(dct):
78 """Validate that the input is a dict with string keys and values.
78 """Validate that the input is a dict with string keys and values.
79
79
80 Raises ValueError if not."""
80 Raises ValueError if not."""
81 for k,v in dct.iteritems():
81 for k,v in dct.iteritems():
82 if not isinstance(k, basestring):
82 if not isinstance(k, basestring):
83 raise ValueError('key %r in dict must be a string' % k)
83 raise ValueError('key %r in dict must be a string' % k)
84 if not isinstance(v, basestring):
84 if not isinstance(v, basestring):
85 raise ValueError('value %r in dict must be a string' % v)
85 raise ValueError('value %r in dict must be a string' % v)
86
86
87
87
88 #-----------------------------------------------------------------------------
88 #-----------------------------------------------------------------------------
89 # ZMQ Socket Channel classes
89 # ZMQ Socket Channel classes
90 #-----------------------------------------------------------------------------
90 #-----------------------------------------------------------------------------
91
91
92 class ZMQSocketChannel(Thread):
92 class ZMQSocketChannel(Thread):
93 """The base class for the channels that use ZMQ sockets."""
93 """The base class for the channels that use ZMQ sockets."""
94 context = None
94 context = None
95 session = None
95 session = None
96 socket = None
96 socket = None
97 ioloop = None
97 ioloop = None
98 stream = None
98 stream = None
99 _address = None
99 _address = None
100 _exiting = False
100 _exiting = False
101
101
102 def __init__(self, context, session, address):
102 def __init__(self, context, session, address):
103 """Create a channel.
103 """Create a channel.
104
104
105 Parameters
105 Parameters
106 ----------
106 ----------
107 context : :class:`zmq.Context`
107 context : :class:`zmq.Context`
108 The ZMQ context to use.
108 The ZMQ context to use.
109 session : :class:`session.Session`
109 session : :class:`session.Session`
110 The session to use.
110 The session to use.
111 address : zmq url
111 address : zmq url
112 Standard (ip, port) tuple that the kernel is listening on.
112 Standard (ip, port) tuple that the kernel is listening on.
113 """
113 """
114 super(ZMQSocketChannel, self).__init__()
114 super(ZMQSocketChannel, self).__init__()
115 self.daemon = True
115 self.daemon = True
116
116
117 self.context = context
117 self.context = context
118 self.session = session
118 self.session = session
119 if isinstance(address, tuple):
119 if isinstance(address, tuple):
120 if address[1] == 0:
120 if address[1] == 0:
121 message = 'The port number for a channel cannot be 0.'
121 message = 'The port number for a channel cannot be 0.'
122 raise InvalidPortNumber(message)
122 raise InvalidPortNumber(message)
123 address = "tcp://%s:%i" % address
123 address = "tcp://%s:%i" % address
124 self._address = address
124 self._address = address
125 atexit.register(self._notice_exit)
125 atexit.register(self._notice_exit)
126
126
127 def _notice_exit(self):
127 def _notice_exit(self):
128 self._exiting = True
128 self._exiting = True
129
129
130 def _run_loop(self):
130 def _run_loop(self):
131 """Run my loop, ignoring EINTR events in the poller"""
131 """Run my loop, ignoring EINTR events in the poller"""
132 while True:
132 while True:
133 try:
133 try:
134 self.ioloop.start()
134 self.ioloop.start()
135 except ZMQError as e:
135 except ZMQError as e:
136 if e.errno == errno.EINTR:
136 if e.errno == errno.EINTR:
137 continue
137 continue
138 else:
138 else:
139 raise
139 raise
140 except Exception:
140 except Exception:
141 if self._exiting:
141 if self._exiting:
142 break
142 break
143 else:
143 else:
144 raise
144 raise
145 else:
145 else:
146 break
146 break
147
147
148 def stop(self):
148 def stop(self):
149 """Stop the channel's event loop and join its thread.
149 """Stop the channel's event loop and join its thread.
150
150
151 This calls :method:`Thread.join` and returns when the thread
151 This calls :method:`Thread.join` and returns when the thread
152 terminates. :class:`RuntimeError` will be raised if
152 terminates. :class:`RuntimeError` will be raised if
153 :method:`self.start` is called again.
153 :method:`self.start` is called again.
154 """
154 """
155 self.join()
155 self.join()
156
156
157 @property
157 @property
158 def address(self):
158 def address(self):
159 """Get the channel's address as a zmq url string.
159 """Get the channel's address as a zmq url string.
160
160
161 These URLS have the form: 'tcp://127.0.0.1:5555'.
161 These URLS have the form: 'tcp://127.0.0.1:5555'.
162 """
162 """
163 return self._address
163 return self._address
164
164
165 def _queue_send(self, msg):
165 def _queue_send(self, msg):
166 """Queue a message to be sent from the IOLoop's thread.
166 """Queue a message to be sent from the IOLoop's thread.
167
167
168 Parameters
168 Parameters
169 ----------
169 ----------
170 msg : message to send
170 msg : message to send
171
171
172 This is threadsafe, as it uses IOLoop.add_callback to give the loop's
172 This is threadsafe, as it uses IOLoop.add_callback to give the loop's
173 thread control of the action.
173 thread control of the action.
174 """
174 """
175 def thread_send():
175 def thread_send():
176 self.session.send(self.stream, msg)
176 self.session.send(self.stream, msg)
177 self.ioloop.add_callback(thread_send)
177 self.ioloop.add_callback(thread_send)
178
178
179 def _handle_recv(self, msg):
179 def _handle_recv(self, msg):
180 """Callback for stream.on_recv.
180 """Callback for stream.on_recv.
181
181
182 Unpacks message, and calls handlers with it.
182 Unpacks message, and calls handlers with it.
183 """
183 """
184 ident,smsg = self.session.feed_identities(msg)
184 ident,smsg = self.session.feed_identities(msg)
185 self.call_handlers(self.session.unserialize(smsg))
185 self.call_handlers(self.session.unserialize(smsg))
186
186
187
187
188
188
189 class ShellChannel(ZMQSocketChannel):
189 class ShellChannel(ZMQSocketChannel):
190 """The shell channel for issuing request/replies to the kernel."""
190 """The shell channel for issuing request/replies to the kernel."""
191
191
192 command_queue = None
192 command_queue = None
193 # flag for whether execute requests should be allowed to call raw_input:
193 # flag for whether execute requests should be allowed to call raw_input:
194 allow_stdin = True
194 allow_stdin = True
195
195
196 def __init__(self, context, session, address):
196 def __init__(self, context, session, address):
197 super(ShellChannel, self).__init__(context, session, address)
197 super(ShellChannel, self).__init__(context, session, address)
198 self.ioloop = ioloop.IOLoop()
198 self.ioloop = ioloop.IOLoop()
199
199
200 def run(self):
200 def run(self):
201 """The thread's main activity. Call start() instead."""
201 """The thread's main activity. Call start() instead."""
202 self.socket = self.context.socket(zmq.DEALER)
202 self.socket = self.context.socket(zmq.DEALER)
203 self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
203 self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
204 self.socket.connect(self.address)
204 self.socket.connect(self.address)
205 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
205 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
206 self.stream.on_recv(self._handle_recv)
206 self.stream.on_recv(self._handle_recv)
207 self._run_loop()
207 self._run_loop()
208 try:
208 try:
209 self.socket.close()
209 self.socket.close()
210 except:
210 except:
211 pass
211 pass
212
212
213 def stop(self):
213 def stop(self):
214 """Stop the channel's event loop and join its thread."""
214 """Stop the channel's event loop and join its thread."""
215 self.ioloop.stop()
215 self.ioloop.stop()
216 super(ShellChannel, self).stop()
216 super(ShellChannel, self).stop()
217
217
218 def call_handlers(self, msg):
218 def call_handlers(self, msg):
219 """This method is called in the ioloop thread when a message arrives.
219 """This method is called in the ioloop thread when a message arrives.
220
220
221 Subclasses should override this method to handle incoming messages.
221 Subclasses should override this method to handle incoming messages.
222 It is important to remember that this method is called in the thread
222 It is important to remember that this method is called in the thread
223 so that some logic must be done to ensure that the application leve
223 so that some logic must be done to ensure that the application leve
224 handlers are called in the application thread.
224 handlers are called in the application thread.
225 """
225 """
226 raise NotImplementedError('call_handlers must be defined in a subclass.')
226 raise NotImplementedError('call_handlers must be defined in a subclass.')
227
227
228 def execute(self, code, silent=False, store_history=True,
228 def execute(self, code, silent=False, store_history=True,
229 user_variables=None, user_expressions=None, allow_stdin=None):
229 user_variables=None, user_expressions=None, allow_stdin=None):
230 """Execute code in the kernel.
230 """Execute code in the kernel.
231
231
232 Parameters
232 Parameters
233 ----------
233 ----------
234 code : str
234 code : str
235 A string of Python code.
235 A string of Python code.
236
236
237 silent : bool, optional (default False)
237 silent : bool, optional (default False)
238 If set, the kernel will execute the code as quietly possible, and
238 If set, the kernel will execute the code as quietly possible, and
239 will force store_history to be False.
239 will force store_history to be False.
240
240
241 store_history : bool, optional (default True)
241 store_history : bool, optional (default True)
242 If set, the kernel will store command history. This is forced
242 If set, the kernel will store command history. This is forced
243 to be False if silent is True.
243 to be False if silent is True.
244
244
245 user_variables : list, optional
245 user_variables : list, optional
246 A list of variable names to pull from the user's namespace. They
246 A list of variable names to pull from the user's namespace. They
247 will come back as a dict with these names as keys and their
247 will come back as a dict with these names as keys and their
248 :func:`repr` as values.
248 :func:`repr` as values.
249
249
250 user_expressions : dict, optional
250 user_expressions : dict, optional
251 A dict mapping names to expressions to be evaluated in the user's
251 A dict mapping names to expressions to be evaluated in the user's
252 dict. The expression values are returned as strings formatted using
252 dict. The expression values are returned as strings formatted using
253 :func:`repr`.
253 :func:`repr`.
254
254
255 allow_stdin : bool, optional (default self.allow_stdin)
255 allow_stdin : bool, optional (default self.allow_stdin)
256 Flag for whether the kernel can send stdin requests to frontends.
256 Flag for whether the kernel can send stdin requests to frontends.
257
257
258 Some frontends (e.g. the Notebook) do not support stdin requests.
258 Some frontends (e.g. the Notebook) do not support stdin requests.
259 If raw_input is called from code executed from such a frontend, a
259 If raw_input is called from code executed from such a frontend, a
260 StdinNotImplementedError will be raised.
260 StdinNotImplementedError will be raised.
261
261
262 Returns
262 Returns
263 -------
263 -------
264 The msg_id of the message sent.
264 The msg_id of the message sent.
265 """
265 """
266 if user_variables is None:
266 if user_variables is None:
267 user_variables = []
267 user_variables = []
268 if user_expressions is None:
268 if user_expressions is None:
269 user_expressions = {}
269 user_expressions = {}
270 if allow_stdin is None:
270 if allow_stdin is None:
271 allow_stdin = self.allow_stdin
271 allow_stdin = self.allow_stdin
272
272
273
273
274 # Don't waste network traffic if inputs are invalid
274 # Don't waste network traffic if inputs are invalid
275 if not isinstance(code, basestring):
275 if not isinstance(code, basestring):
276 raise ValueError('code %r must be a string' % code)
276 raise ValueError('code %r must be a string' % code)
277 validate_string_list(user_variables)
277 validate_string_list(user_variables)
278 validate_string_dict(user_expressions)
278 validate_string_dict(user_expressions)
279
279
280 # Create class for content/msg creation. Related to, but possibly
280 # Create class for content/msg creation. Related to, but possibly
281 # not in Session.
281 # not in Session.
282 content = dict(code=code, silent=silent, store_history=store_history,
282 content = dict(code=code, silent=silent, store_history=store_history,
283 user_variables=user_variables,
283 user_variables=user_variables,
284 user_expressions=user_expressions,
284 user_expressions=user_expressions,
285 allow_stdin=allow_stdin,
285 allow_stdin=allow_stdin,
286 )
286 )
287 msg = self.session.msg('execute_request', content)
287 msg = self.session.msg('execute_request', content)
288 self._queue_send(msg)
288 self._queue_send(msg)
289 return msg['header']['msg_id']
289 return msg['header']['msg_id']
290
290
291 def complete(self, text, line, cursor_pos, block=None):
291 def complete(self, text, line, cursor_pos, block=None):
292 """Tab complete text in the kernel's namespace.
292 """Tab complete text in the kernel's namespace.
293
293
294 Parameters
294 Parameters
295 ----------
295 ----------
296 text : str
296 text : str
297 The text to complete.
297 The text to complete.
298 line : str
298 line : str
299 The full line of text that is the surrounding context for the
299 The full line of text that is the surrounding context for the
300 text to complete.
300 text to complete.
301 cursor_pos : int
301 cursor_pos : int
302 The position of the cursor in the line where the completion was
302 The position of the cursor in the line where the completion was
303 requested.
303 requested.
304 block : str, optional
304 block : str, optional
305 The full block of code in which the completion is being requested.
305 The full block of code in which the completion is being requested.
306
306
307 Returns
307 Returns
308 -------
308 -------
309 The msg_id of the message sent.
309 The msg_id of the message sent.
310 """
310 """
311 content = dict(text=text, line=line, block=block, cursor_pos=cursor_pos)
311 content = dict(text=text, line=line, block=block, cursor_pos=cursor_pos)
312 msg = self.session.msg('complete_request', content)
312 msg = self.session.msg('complete_request', content)
313 self._queue_send(msg)
313 self._queue_send(msg)
314 return msg['header']['msg_id']
314 return msg['header']['msg_id']
315
315
316 def object_info(self, oname, detail_level=0):
316 def object_info(self, oname, detail_level=0):
317 """Get metadata information about an object in the kernel's namespace.
317 """Get metadata information about an object in the kernel's namespace.
318
318
319 Parameters
319 Parameters
320 ----------
320 ----------
321 oname : str
321 oname : str
322 A string specifying the object name.
322 A string specifying the object name.
323 detail_level : int, optional
323 detail_level : int, optional
324 The level of detail for the introspection (0-2)
324 The level of detail for the introspection (0-2)
325
325
326 Returns
326 Returns
327 -------
327 -------
328 The msg_id of the message sent.
328 The msg_id of the message sent.
329 """
329 """
330 content = dict(oname=oname, detail_level=detail_level)
330 content = dict(oname=oname, detail_level=detail_level)
331 msg = self.session.msg('object_info_request', content)
331 msg = self.session.msg('object_info_request', content)
332 self._queue_send(msg)
332 self._queue_send(msg)
333 return msg['header']['msg_id']
333 return msg['header']['msg_id']
334
334
335 def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
335 def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
336 """Get entries from the kernel's history list.
336 """Get entries from the kernel's history list.
337
337
338 Parameters
338 Parameters
339 ----------
339 ----------
340 raw : bool
340 raw : bool
341 If True, return the raw input.
341 If True, return the raw input.
342 output : bool
342 output : bool
343 If True, then return the output as well.
343 If True, then return the output as well.
344 hist_access_type : str
344 hist_access_type : str
345 'range' (fill in session, start and stop params), 'tail' (fill in n)
345 'range' (fill in session, start and stop params), 'tail' (fill in n)
346 or 'search' (fill in pattern param).
346 or 'search' (fill in pattern param).
347
347
348 session : int
348 session : int
349 For a range request, the session from which to get lines. Session
349 For a range request, the session from which to get lines. Session
350 numbers are positive integers; negative ones count back from the
350 numbers are positive integers; negative ones count back from the
351 current session.
351 current session.
352 start : int
352 start : int
353 The first line number of a history range.
353 The first line number of a history range.
354 stop : int
354 stop : int
355 The final (excluded) line number of a history range.
355 The final (excluded) line number of a history range.
356
356
357 n : int
357 n : int
358 The number of lines of history to get for a tail request.
358 The number of lines of history to get for a tail request.
359
359
360 pattern : str
360 pattern : str
361 The glob-syntax pattern for a search request.
361 The glob-syntax pattern for a search request.
362
362
363 Returns
363 Returns
364 -------
364 -------
365 The msg_id of the message sent.
365 The msg_id of the message sent.
366 """
366 """
367 content = dict(raw=raw, output=output, hist_access_type=hist_access_type,
367 content = dict(raw=raw, output=output, hist_access_type=hist_access_type,
368 **kwargs)
368 **kwargs)
369 msg = self.session.msg('history_request', content)
369 msg = self.session.msg('history_request', content)
370 self._queue_send(msg)
370 self._queue_send(msg)
371 return msg['header']['msg_id']
371 return msg['header']['msg_id']
372
372
373 def kernel_info(self):
373 def kernel_info(self):
374 """Request kernel info."""
374 """Request kernel info."""
375 msg = self.session.msg('kernel_info_request')
375 msg = self.session.msg('kernel_info_request')
376 self._queue_send(msg)
376 self._queue_send(msg)
377 return msg['header']['msg_id']
377 return msg['header']['msg_id']
378
378
379 def shutdown(self, restart=False):
379 def shutdown(self, restart=False):
380 """Request an immediate kernel shutdown.
380 """Request an immediate kernel shutdown.
381
381
382 Upon receipt of the (empty) reply, client code can safely assume that
382 Upon receipt of the (empty) reply, client code can safely assume that
383 the kernel has shut down and it's safe to forcefully terminate it if
383 the kernel has shut down and it's safe to forcefully terminate it if
384 it's still alive.
384 it's still alive.
385
385
386 The kernel will send the reply via a function registered with Python's
386 The kernel will send the reply via a function registered with Python's
387 atexit module, ensuring it's truly done as the kernel is done with all
387 atexit module, ensuring it's truly done as the kernel is done with all
388 normal operation.
388 normal operation.
389 """
389 """
390 # Send quit message to kernel. Once we implement kernel-side setattr,
390 # Send quit message to kernel. Once we implement kernel-side setattr,
391 # this should probably be done that way, but for now this will do.
391 # this should probably be done that way, but for now this will do.
392 msg = self.session.msg('shutdown_request', {'restart':restart})
392 msg = self.session.msg('shutdown_request', {'restart':restart})
393 self._queue_send(msg)
393 self._queue_send(msg)
394 return msg['header']['msg_id']
394 return msg['header']['msg_id']
395
395
396
396
397
397
398 class IOPubChannel(ZMQSocketChannel):
398 class IOPubChannel(ZMQSocketChannel):
399 """The iopub channel which listens for messages that the kernel publishes.
399 """The iopub channel which listens for messages that the kernel publishes.
400
400
401 This channel is where all output is published to frontends.
401 This channel is where all output is published to frontends.
402 """
402 """
403
403
404 def __init__(self, context, session, address):
404 def __init__(self, context, session, address):
405 super(IOPubChannel, self).__init__(context, session, address)
405 super(IOPubChannel, self).__init__(context, session, address)
406 self.ioloop = ioloop.IOLoop()
406 self.ioloop = ioloop.IOLoop()
407
407
408 def run(self):
408 def run(self):
409 """The thread's main activity. Call start() instead."""
409 """The thread's main activity. Call start() instead."""
410 self.socket = self.context.socket(zmq.SUB)
410 self.socket = self.context.socket(zmq.SUB)
411 self.socket.setsockopt(zmq.SUBSCRIBE,b'')
411 self.socket.setsockopt(zmq.SUBSCRIBE,b'')
412 self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
412 self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
413 self.socket.connect(self.address)
413 self.socket.connect(self.address)
414 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
414 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
415 self.stream.on_recv(self._handle_recv)
415 self.stream.on_recv(self._handle_recv)
416 self._run_loop()
416 self._run_loop()
417 try:
417 try:
418 self.socket.close()
418 self.socket.close()
419 except:
419 except:
420 pass
420 pass
421
421
422 def stop(self):
422 def stop(self):
423 """Stop the channel's event loop and join its thread."""
423 """Stop the channel's event loop and join its thread."""
424 self.ioloop.stop()
424 self.ioloop.stop()
425 super(IOPubChannel, self).stop()
425 super(IOPubChannel, self).stop()
426
426
427 def call_handlers(self, msg):
427 def call_handlers(self, msg):
428 """This method is called in the ioloop thread when a message arrives.
428 """This method is called in the ioloop thread when a message arrives.
429
429
430 Subclasses should override this method to handle incoming messages.
430 Subclasses should override this method to handle incoming messages.
431 It is important to remember that this method is called in the thread
431 It is important to remember that this method is called in the thread
432 so that some logic must be done to ensure that the application leve
432 so that some logic must be done to ensure that the application leve
433 handlers are called in the application thread.
433 handlers are called in the application thread.
434 """
434 """
435 raise NotImplementedError('call_handlers must be defined in a subclass.')
435 raise NotImplementedError('call_handlers must be defined in a subclass.')
436
436
437 def flush(self, timeout=1.0):
437 def flush(self, timeout=1.0):
438 """Immediately processes all pending messages on the iopub channel.
438 """Immediately processes all pending messages on the iopub channel.
439
439
440 Callers should use this method to ensure that :method:`call_handlers`
440 Callers should use this method to ensure that :method:`call_handlers`
441 has been called for all messages that have been received on the
441 has been called for all messages that have been received on the
442 0MQ SUB socket of this channel.
442 0MQ SUB socket of this channel.
443
443
444 This method is thread safe.
444 This method is thread safe.
445
445
446 Parameters
446 Parameters
447 ----------
447 ----------
448 timeout : float, optional
448 timeout : float, optional
449 The maximum amount of time to spend flushing, in seconds. The
449 The maximum amount of time to spend flushing, in seconds. The
450 default is one second.
450 default is one second.
451 """
451 """
452 # We do the IOLoop callback process twice to ensure that the IOLoop
452 # We do the IOLoop callback process twice to ensure that the IOLoop
453 # gets to perform at least one full poll.
453 # gets to perform at least one full poll.
454 stop_time = time.time() + timeout
454 stop_time = time.time() + timeout
455 for i in xrange(2):
455 for i in xrange(2):
456 self._flushed = False
456 self._flushed = False
457 self.ioloop.add_callback(self._flush)
457 self.ioloop.add_callback(self._flush)
458 while not self._flushed and time.time() < stop_time:
458 while not self._flushed and time.time() < stop_time:
459 time.sleep(0.01)
459 time.sleep(0.01)
460
460
461 def _flush(self):
461 def _flush(self):
462 """Callback for :method:`self.flush`."""
462 """Callback for :method:`self.flush`."""
463 self.stream.flush()
463 self.stream.flush()
464 self._flushed = True
464 self._flushed = True
465
465
466
466
467 class StdInChannel(ZMQSocketChannel):
467 class StdInChannel(ZMQSocketChannel):
468 """The stdin channel to handle raw_input requests that the kernel makes."""
468 """The stdin channel to handle raw_input requests that the kernel makes."""
469
469
470 msg_queue = None
470 msg_queue = None
471
471
472 def __init__(self, context, session, address):
472 def __init__(self, context, session, address):
473 super(StdInChannel, self).__init__(context, session, address)
473 super(StdInChannel, self).__init__(context, session, address)
474 self.ioloop = ioloop.IOLoop()
474 self.ioloop = ioloop.IOLoop()
475
475
476 def run(self):
476 def run(self):
477 """The thread's main activity. Call start() instead."""
477 """The thread's main activity. Call start() instead."""
478 self.socket = self.context.socket(zmq.DEALER)
478 self.socket = self.context.socket(zmq.DEALER)
479 self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
479 self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
480 self.socket.connect(self.address)
480 self.socket.connect(self.address)
481 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
481 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
482 self.stream.on_recv(self._handle_recv)
482 self.stream.on_recv(self._handle_recv)
483 self._run_loop()
483 self._run_loop()
484 try:
484 try:
485 self.socket.close()
485 self.socket.close()
486 except:
486 except:
487 pass
487 pass
488
488
489 def stop(self):
489 def stop(self):
490 """Stop the channel's event loop and join its thread."""
490 """Stop the channel's event loop and join its thread."""
491 self.ioloop.stop()
491 self.ioloop.stop()
492 super(StdInChannel, self).stop()
492 super(StdInChannel, self).stop()
493
493
494 def call_handlers(self, msg):
494 def call_handlers(self, msg):
495 """This method is called in the ioloop thread when a message arrives.
495 """This method is called in the ioloop thread when a message arrives.
496
496
497 Subclasses should override this method to handle incoming messages.
497 Subclasses should override this method to handle incoming messages.
498 It is important to remember that this method is called in the thread
498 It is important to remember that this method is called in the thread
499 so that some logic must be done to ensure that the application leve
499 so that some logic must be done to ensure that the application leve
500 handlers are called in the application thread.
500 handlers are called in the application thread.
501 """
501 """
502 raise NotImplementedError('call_handlers must be defined in a subclass.')
502 raise NotImplementedError('call_handlers must be defined in a subclass.')
503
503
504 def input(self, string):
504 def input(self, string):
505 """Send a string of raw input to the kernel."""
505 """Send a string of raw input to the kernel."""
506 content = dict(value=string)
506 content = dict(value=string)
507 msg = self.session.msg('input_reply', content)
507 msg = self.session.msg('input_reply', content)
508 self._queue_send(msg)
508 self._queue_send(msg)
509
509
510
510
511 class HBChannel(ZMQSocketChannel):
511 class HBChannel(ZMQSocketChannel):
512 """The heartbeat channel which monitors the kernel heartbeat.
512 """The heartbeat channel which monitors the kernel heartbeat.
513
513
514 Note that the heartbeat channel is paused by default. As long as you start
514 Note that the heartbeat channel is paused by default. As long as you start
515 this channel, the kernel manager will ensure that it is paused and un-paused
515 this channel, the kernel manager will ensure that it is paused and un-paused
516 as appropriate.
516 as appropriate.
517 """
517 """
518
518
519 time_to_dead = 3.0
519 time_to_dead = 3.0
520 socket = None
520 socket = None
521 poller = None
521 poller = None
522 _running = None
522 _running = None
523 _pause = None
523 _pause = None
524 _beating = None
524 _beating = None
525
525
526 def __init__(self, context, session, address):
526 def __init__(self, context, session, address):
527 super(HBChannel, self).__init__(context, session, address)
527 super(HBChannel, self).__init__(context, session, address)
528 self._running = False
528 self._running = False
529 self._pause =True
529 self._pause =True
530 self.poller = zmq.Poller()
530 self.poller = zmq.Poller()
531
531
532 def _create_socket(self):
532 def _create_socket(self):
533 if self.socket is not None:
533 if self.socket is not None:
534 # close previous socket, before opening a new one
534 # close previous socket, before opening a new one
535 self.poller.unregister(self.socket)
535 self.poller.unregister(self.socket)
536 self.socket.close()
536 self.socket.close()
537 self.socket = self.context.socket(zmq.REQ)
537 self.socket = self.context.socket(zmq.REQ)
538 self.socket.setsockopt(zmq.LINGER, 0)
538 self.socket.setsockopt(zmq.LINGER, 0)
539 self.socket.connect(self.address)
539 self.socket.connect(self.address)
540
540
541 self.poller.register(self.socket, zmq.POLLIN)
541 self.poller.register(self.socket, zmq.POLLIN)
542
542
543 def _poll(self, start_time):
543 def _poll(self, start_time):
544 """poll for heartbeat replies until we reach self.time_to_dead.
544 """poll for heartbeat replies until we reach self.time_to_dead.
545
545
546 Ignores interrupts, and returns the result of poll(), which
546 Ignores interrupts, and returns the result of poll(), which
547 will be an empty list if no messages arrived before the timeout,
547 will be an empty list if no messages arrived before the timeout,
548 or the event tuple if there is a message to receive.
548 or the event tuple if there is a message to receive.
549 """
549 """
550
550
551 until_dead = self.time_to_dead - (time.time() - start_time)
551 until_dead = self.time_to_dead - (time.time() - start_time)
552 # ensure poll at least once
552 # ensure poll at least once
553 until_dead = max(until_dead, 1e-3)
553 until_dead = max(until_dead, 1e-3)
554 events = []
554 events = []
555 while True:
555 while True:
556 try:
556 try:
557 events = self.poller.poll(1000 * until_dead)
557 events = self.poller.poll(1000 * until_dead)
558 except ZMQError as e:
558 except ZMQError as e:
559 if e.errno == errno.EINTR:
559 if e.errno == errno.EINTR:
560 # ignore interrupts during heartbeat
560 # ignore interrupts during heartbeat
561 # this may never actually happen
561 # this may never actually happen
562 until_dead = self.time_to_dead - (time.time() - start_time)
562 until_dead = self.time_to_dead - (time.time() - start_time)
563 until_dead = max(until_dead, 1e-3)
563 until_dead = max(until_dead, 1e-3)
564 pass
564 pass
565 else:
565 else:
566 raise
566 raise
567 except Exception:
567 except Exception:
568 if self._exiting:
568 if self._exiting:
569 break
569 break
570 else:
570 else:
571 raise
571 raise
572 else:
572 else:
573 break
573 break
574 return events
574 return events
575
575
576 def run(self):
576 def run(self):
577 """The thread's main activity. Call start() instead."""
577 """The thread's main activity. Call start() instead."""
578 self._create_socket()
578 self._create_socket()
579 self._running = True
579 self._running = True
580 self._beating = True
580 self._beating = True
581
581
582 while self._running:
582 while self._running:
583 if self._pause:
583 if self._pause:
584 # just sleep, and skip the rest of the loop
584 # just sleep, and skip the rest of the loop
585 time.sleep(self.time_to_dead)
585 time.sleep(self.time_to_dead)
586 continue
586 continue
587
587
588 since_last_heartbeat = 0.0
588 since_last_heartbeat = 0.0
589 # io.rprint('Ping from HB channel') # dbg
589 # io.rprint('Ping from HB channel') # dbg
590 # no need to catch EFSM here, because the previous event was
590 # no need to catch EFSM here, because the previous event was
591 # either a recv or connect, which cannot be followed by EFSM
591 # either a recv or connect, which cannot be followed by EFSM
592 self.socket.send(b'ping')
592 self.socket.send(b'ping')
593 request_time = time.time()
593 request_time = time.time()
594 ready = self._poll(request_time)
594 ready = self._poll(request_time)
595 if ready:
595 if ready:
596 self._beating = True
596 self._beating = True
597 # the poll above guarantees we have something to recv
597 # the poll above guarantees we have something to recv
598 self.socket.recv()
598 self.socket.recv()
599 # sleep the remainder of the cycle
599 # sleep the remainder of the cycle
600 remainder = self.time_to_dead - (time.time() - request_time)
600 remainder = self.time_to_dead - (time.time() - request_time)
601 if remainder > 0:
601 if remainder > 0:
602 time.sleep(remainder)
602 time.sleep(remainder)
603 continue
603 continue
604 else:
604 else:
605 # nothing was received within the time limit, signal heart failure
605 # nothing was received within the time limit, signal heart failure
606 self._beating = False
606 self._beating = False
607 since_last_heartbeat = time.time() - request_time
607 since_last_heartbeat = time.time() - request_time
608 self.call_handlers(since_last_heartbeat)
608 self.call_handlers(since_last_heartbeat)
609 # and close/reopen the socket, because the REQ/REP cycle has been broken
609 # and close/reopen the socket, because the REQ/REP cycle has been broken
610 self._create_socket()
610 self._create_socket()
611 continue
611 continue
612 try:
612 try:
613 self.socket.close()
613 self.socket.close()
614 except:
614 except:
615 pass
615 pass
616
616
617 def pause(self):
617 def pause(self):
618 """Pause the heartbeat."""
618 """Pause the heartbeat."""
619 self._pause = True
619 self._pause = True
620
620
621 def unpause(self):
621 def unpause(self):
622 """Unpause the heartbeat."""
622 """Unpause the heartbeat."""
623 self._pause = False
623 self._pause = False
624
624
625 def is_beating(self):
625 def is_beating(self):
626 """Is the heartbeat running and responsive (and not paused)."""
626 """Is the heartbeat running and responsive (and not paused)."""
627 if self.is_alive() and not self._pause and self._beating:
627 if self.is_alive() and not self._pause and self._beating:
628 return True
628 return True
629 else:
629 else:
630 return False
630 return False
631
631
632 def stop(self):
632 def stop(self):
633 """Stop the channel's event loop and join its thread."""
633 """Stop the channel's event loop and join its thread."""
634 self._running = False
634 self._running = False
635 super(HBChannel, self).stop()
635 super(HBChannel, self).stop()
636
636
637 def call_handlers(self, since_last_heartbeat):
637 def call_handlers(self, since_last_heartbeat):
638 """This method is called in the ioloop thread when a message arrives.
638 """This method is called in the ioloop thread when a message arrives.
639
639
640 Subclasses should override this method to handle incoming messages.
640 Subclasses should override this method to handle incoming messages.
641 It is important to remember that this method is called in the thread
641 It is important to remember that this method is called in the thread
642 so that some logic must be done to ensure that the application level
642 so that some logic must be done to ensure that the application level
643 handlers are called in the application thread.
643 handlers are called in the application thread.
644 """
644 """
645 raise NotImplementedError('call_handlers must be defined in a subclass.')
645 raise NotImplementedError('call_handlers must be defined in a subclass.')
646
646
647
647
648 #-----------------------------------------------------------------------------
648 #-----------------------------------------------------------------------------
649 # Main kernel manager class
649 # Main kernel manager class
650 #-----------------------------------------------------------------------------
650 #-----------------------------------------------------------------------------
651
651
652 class KernelManager(Configurable):
652 class KernelManager(Configurable):
653 """Manages a single kernel on this host along with its channels.
653 """Manages a single kernel on this host along with its channels.
654
654
655 There are four channels associated with each kernel:
655 There are four channels associated with each kernel:
656
656
657 * shell: for request/reply calls to the kernel.
657 * shell: for request/reply calls to the kernel.
658 * iopub: for the kernel to publish results to frontends.
658 * iopub: for the kernel to publish results to frontends.
659 * hb: for monitoring the kernel's heartbeat.
659 * hb: for monitoring the kernel's heartbeat.
660 * stdin: for frontends to reply to raw_input calls in the kernel.
660 * stdin: for frontends to reply to raw_input calls in the kernel.
661
661
662 The usage of the channels that this class manages is optional. It is
662 The usage of the channels that this class manages is optional. It is
663 entirely possible to connect to the kernels directly using ZeroMQ
663 entirely possible to connect to the kernels directly using ZeroMQ
664 sockets. These channels are useful primarily for talking to a kernel
664 sockets. These channels are useful primarily for talking to a kernel
665 whose :class:`KernelManager` is in the same process.
665 whose :class:`KernelManager` is in the same process.
666
666
667 This version manages kernels started using Popen.
667 This version manages kernels started using Popen.
668 """
668 """
669 # The PyZMQ Context to use for communication with the kernel.
669 # The PyZMQ Context to use for communication with the kernel.
670 context = Instance(zmq.Context)
670 context = Instance(zmq.Context)
671 def _context_default(self):
671 def _context_default(self):
672 return zmq.Context.instance()
672 return zmq.Context.instance()
673
673
674 # The Session to use for communication with the kernel.
674 # The Session to use for communication with the kernel.
675 session = Instance(Session)
675 session = Instance(Session)
676 def _session_default(self):
676 def _session_default(self):
677 return Session(config=self.config)
677 return Session(config=self.config)
678
678
679 # The kernel process with which the KernelManager is communicating.
679 # The kernel process with which the KernelManager is communicating.
680 kernel = Instance(Popen)
680 kernel = Instance(Popen)
681
681
682 # The addresses for the communication channels.
682 # The addresses for the communication channels.
683 connection_file = Unicode('')
683 connection_file = Unicode('')
684
684
685 transport = CaselessStrEnum(['tcp', 'ipc'], default_value='tcp', config=True)
685 transport = CaselessStrEnum(['tcp', 'ipc'], default_value='tcp', config=True)
686
686
687 ip = Unicode(LOCALHOST, config=True)
687 ip = Unicode(LOCALHOST, config=True,
688 help="""Set the kernel\'s IP address [default localhost].
689 If the IP address is something other than localhost, then
690 Consoles on other machines will be able to connect
691 to the Kernel, so be careful!"""
692 )
693 def _ip_default(self):
694 if self.transport == 'ipc':
695 if self.connection_file:
696 return os.path.splitext(self.connection_file)[0] + '-ipc'
697 else:
698 return 'kernel-ipc'
699 else:
700 return LOCALHOST
688 def _ip_changed(self, name, old, new):
701 def _ip_changed(self, name, old, new):
689 if new == '*':
702 if new == '*':
690 self.ip = '0.0.0.0'
703 self.ip = '0.0.0.0'
691 shell_port = Integer(0)
704 shell_port = Integer(0)
692 iopub_port = Integer(0)
705 iopub_port = Integer(0)
693 stdin_port = Integer(0)
706 stdin_port = Integer(0)
694 hb_port = Integer(0)
707 hb_port = Integer(0)
695
708
696 # The classes to use for the various channels.
709 # The classes to use for the various channels.
697 shell_channel_class = Type(ShellChannel)
710 shell_channel_class = Type(ShellChannel)
698 iopub_channel_class = Type(IOPubChannel)
711 iopub_channel_class = Type(IOPubChannel)
699 stdin_channel_class = Type(StdInChannel)
712 stdin_channel_class = Type(StdInChannel)
700 hb_channel_class = Type(HBChannel)
713 hb_channel_class = Type(HBChannel)
701
714
702 # Protected traits.
715 # Protected traits.
703 _launch_args = Any
716 _launch_args = Any
704 _shell_channel = Any
717 _shell_channel = Any
705 _iopub_channel = Any
718 _iopub_channel = Any
706 _stdin_channel = Any
719 _stdin_channel = Any
707 _hb_channel = Any
720 _hb_channel = Any
708 _connection_file_written=Bool(False)
721 _connection_file_written=Bool(False)
709
722
710 def __del__(self):
723 def __del__(self):
711 self.cleanup_connection_file()
724 self.cleanup_connection_file()
712
725
713 #--------------------------------------------------------------------------
726 #--------------------------------------------------------------------------
714 # Channel management methods:
727 # Channel management methods:
715 #--------------------------------------------------------------------------
728 #--------------------------------------------------------------------------
716
729
717 def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
730 def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
718 """Starts the channels for this kernel.
731 """Starts the channels for this kernel.
719
732
720 This will create the channels if they do not exist and then start
733 This will create the channels if they do not exist and then start
721 them (their activity runs in a thread). If port numbers of 0 are
734 them (their activity runs in a thread). If port numbers of 0 are
722 being used (random ports) then you must first call
735 being used (random ports) then you must first call
723 :method:`start_kernel`. If the channels have been stopped and you
736 :method:`start_kernel`. If the channels have been stopped and you
724 call this, :class:`RuntimeError` will be raised.
737 call this, :class:`RuntimeError` will be raised.
725 """
738 """
726 if shell:
739 if shell:
727 self.shell_channel.start()
740 self.shell_channel.start()
728 if iopub:
741 if iopub:
729 self.iopub_channel.start()
742 self.iopub_channel.start()
730 if stdin:
743 if stdin:
731 self.stdin_channel.start()
744 self.stdin_channel.start()
732 self.shell_channel.allow_stdin = True
745 self.shell_channel.allow_stdin = True
733 else:
746 else:
734 self.shell_channel.allow_stdin = False
747 self.shell_channel.allow_stdin = False
735 if hb:
748 if hb:
736 self.hb_channel.start()
749 self.hb_channel.start()
737
750
738 def stop_channels(self):
751 def stop_channels(self):
739 """Stops all the running channels for this kernel.
752 """Stops all the running channels for this kernel.
740
753
741 This stops their event loops and joins their threads.
754 This stops their event loops and joins their threads.
742 """
755 """
743 if self.shell_channel.is_alive():
756 if self.shell_channel.is_alive():
744 self.shell_channel.stop()
757 self.shell_channel.stop()
745 if self.iopub_channel.is_alive():
758 if self.iopub_channel.is_alive():
746 self.iopub_channel.stop()
759 self.iopub_channel.stop()
747 if self.stdin_channel.is_alive():
760 if self.stdin_channel.is_alive():
748 self.stdin_channel.stop()
761 self.stdin_channel.stop()
749 if self.hb_channel.is_alive():
762 if self.hb_channel.is_alive():
750 self.hb_channel.stop()
763 self.hb_channel.stop()
751
764
752 @property
765 @property
753 def channels_running(self):
766 def channels_running(self):
754 """Are any of the channels created and running?"""
767 """Are any of the channels created and running?"""
755 return (self.shell_channel.is_alive() or self.iopub_channel.is_alive() or
768 return (self.shell_channel.is_alive() or self.iopub_channel.is_alive() or
756 self.stdin_channel.is_alive() or self.hb_channel.is_alive())
769 self.stdin_channel.is_alive() or self.hb_channel.is_alive())
757
770
758 def _make_url(self, port):
771 def _make_url(self, port):
759 """Make a zmq url with a port.
772 """Make a zmq url with a port.
760
773
761 There are two cases that this handles:
774 There are two cases that this handles:
762
775
763 * tcp: tcp://ip:port
776 * tcp: tcp://ip:port
764 * ipc: ipc://ip-port
777 * ipc: ipc://ip-port
765 """
778 """
766 if self.transport == 'tcp':
779 if self.transport == 'tcp':
767 return "tcp://%s:%i" % (self.ip, port)
780 return "tcp://%s:%i" % (self.ip, port)
768 else:
781 else:
769 return "%s://%s-%s" % (self.transport, self.ip, port)
782 return "%s://%s-%s" % (self.transport, self.ip, port)
770
783
771 @property
784 @property
772 def shell_channel(self):
785 def shell_channel(self):
773 """Get the shell channel object for this kernel."""
786 """Get the shell channel object for this kernel."""
774 if self._shell_channel is None:
787 if self._shell_channel is None:
775 self._shell_channel = self.shell_channel_class(
788 self._shell_channel = self.shell_channel_class(
776 self.context, self.session, self._make_url(self.shell_port)
789 self.context, self.session, self._make_url(self.shell_port)
777 )
790 )
778 return self._shell_channel
791 return self._shell_channel
779
792
780 @property
793 @property
781 def iopub_channel(self):
794 def iopub_channel(self):
782 """Get the iopub channel object for this kernel."""
795 """Get the iopub channel object for this kernel."""
783 if self._iopub_channel is None:
796 if self._iopub_channel is None:
784 self._iopub_channel = self.iopub_channel_class(
797 self._iopub_channel = self.iopub_channel_class(
785 self.context, self.session, self._make_url(self.iopub_port)
798 self.context, self.session, self._make_url(self.iopub_port)
786 )
799 )
787 return self._iopub_channel
800 return self._iopub_channel
788
801
789 @property
802 @property
790 def stdin_channel(self):
803 def stdin_channel(self):
791 """Get the stdin channel object for this kernel."""
804 """Get the stdin channel object for this kernel."""
792 if self._stdin_channel is None:
805 if self._stdin_channel is None:
793 self._stdin_channel = self.stdin_channel_class(
806 self._stdin_channel = self.stdin_channel_class(
794 self.context, self.session, self._make_url(self.stdin_port)
807 self.context, self.session, self._make_url(self.stdin_port)
795 )
808 )
796 return self._stdin_channel
809 return self._stdin_channel
797
810
798 @property
811 @property
799 def hb_channel(self):
812 def hb_channel(self):
800 """Get the hb channel object for this kernel."""
813 """Get the hb channel object for this kernel."""
801 if self._hb_channel is None:
814 if self._hb_channel is None:
802 self._hb_channel = self.hb_channel_class(
815 self._hb_channel = self.hb_channel_class(
803 self.context, self.session, self._make_url(self.hb_port)
816 self.context, self.session, self._make_url(self.hb_port)
804 )
817 )
805 return self._hb_channel
818 return self._hb_channel
806
819
807 #--------------------------------------------------------------------------
820 #--------------------------------------------------------------------------
808 # Connection and ipc file management
821 # Connection and ipc file management
809 #--------------------------------------------------------------------------
822 #--------------------------------------------------------------------------
810
823
811 def cleanup_connection_file(self):
824 def cleanup_connection_file(self):
812 """Cleanup connection file *if we wrote it*
825 """Cleanup connection file *if we wrote it*
813
826
814 Will not raise if the connection file was already removed somehow.
827 Will not raise if the connection file was already removed somehow.
815 """
828 """
816 if self._connection_file_written:
829 if self._connection_file_written:
817 # cleanup connection files on full shutdown of kernel we started
830 # cleanup connection files on full shutdown of kernel we started
818 self._connection_file_written = False
831 self._connection_file_written = False
819 try:
832 try:
820 os.remove(self.connection_file)
833 os.remove(self.connection_file)
821 except (IOError, OSError):
834 except (IOError, OSError):
822 pass
835 pass
823
836
824 def cleanup_ipc_files(self):
837 def cleanup_ipc_files(self):
825 """Cleanup ipc files if we wrote them."""
838 """Cleanup ipc files if we wrote them."""
826 if self.transport != 'ipc':
839 if self.transport != 'ipc':
827 return
840 return
828 for port in (self.shell_port, self.iopub_port, self.stdin_port, self.hb_port):
841 for port in (self.shell_port, self.iopub_port, self.stdin_port, self.hb_port):
829 ipcfile = "%s-%i" % (self.ip, port)
842 ipcfile = "%s-%i" % (self.ip, port)
830 try:
843 try:
831 os.remove(ipcfile)
844 os.remove(ipcfile)
832 except (IOError, OSError):
845 except (IOError, OSError):
833 pass
846 pass
834
847
835 def load_connection_file(self):
848 def load_connection_file(self):
836 """Load connection info from JSON dict in self.connection_file."""
849 """Load connection info from JSON dict in self.connection_file."""
837 with open(self.connection_file) as f:
850 with open(self.connection_file) as f:
838 cfg = json.loads(f.read())
851 cfg = json.loads(f.read())
839
852
840 from pprint import pprint
853 from pprint import pprint
841 pprint(cfg)
854 pprint(cfg)
842 self.transport = cfg.get('transport', 'tcp')
855 self.transport = cfg.get('transport', 'tcp')
843 self.ip = cfg['ip']
856 self.ip = cfg['ip']
844 self.shell_port = cfg['shell_port']
857 self.shell_port = cfg['shell_port']
845 self.stdin_port = cfg['stdin_port']
858 self.stdin_port = cfg['stdin_port']
846 self.iopub_port = cfg['iopub_port']
859 self.iopub_port = cfg['iopub_port']
847 self.hb_port = cfg['hb_port']
860 self.hb_port = cfg['hb_port']
848 self.session.key = str_to_bytes(cfg['key'])
861 self.session.key = str_to_bytes(cfg['key'])
849
862
850 def write_connection_file(self):
863 def write_connection_file(self):
851 """Write connection info to JSON dict in self.connection_file."""
864 """Write connection info to JSON dict in self.connection_file."""
852 if self._connection_file_written:
865 if self._connection_file_written:
853 return
866 return
854 self.connection_file,cfg = write_connection_file(self.connection_file,
867 self.connection_file,cfg = write_connection_file(self.connection_file,
855 transport=self.transport, ip=self.ip, key=self.session.key,
868 transport=self.transport, ip=self.ip, key=self.session.key,
856 stdin_port=self.stdin_port, iopub_port=self.iopub_port,
869 stdin_port=self.stdin_port, iopub_port=self.iopub_port,
857 shell_port=self.shell_port, hb_port=self.hb_port)
870 shell_port=self.shell_port, hb_port=self.hb_port)
858 # write_connection_file also sets default ports:
871 # write_connection_file also sets default ports:
859 self.shell_port = cfg['shell_port']
872 self.shell_port = cfg['shell_port']
860 self.stdin_port = cfg['stdin_port']
873 self.stdin_port = cfg['stdin_port']
861 self.iopub_port = cfg['iopub_port']
874 self.iopub_port = cfg['iopub_port']
862 self.hb_port = cfg['hb_port']
875 self.hb_port = cfg['hb_port']
863
876
864 self._connection_file_written = True
877 self._connection_file_written = True
865
878
866 #--------------------------------------------------------------------------
879 #--------------------------------------------------------------------------
867 # Kernel management
880 # Kernel management
868 #--------------------------------------------------------------------------
881 #--------------------------------------------------------------------------
869
882
870 def start_kernel(self, **kw):
883 def start_kernel(self, **kw):
871 """Starts a kernel on this host in a separate process.
884 """Starts a kernel on this host in a separate process.
872
885
873 If random ports (port=0) are being used, this method must be called
886 If random ports (port=0) are being used, this method must be called
874 before the channels are created.
887 before the channels are created.
875
888
876 Parameters:
889 Parameters:
877 -----------
890 -----------
878 launcher : callable, optional (default None)
891 launcher : callable, optional (default None)
879 A custom function for launching the kernel process (generally a
892 A custom function for launching the kernel process (generally a
880 wrapper around ``entry_point.base_launch_kernel``). In most cases,
893 wrapper around ``entry_point.base_launch_kernel``). In most cases,
881 it should not be necessary to use this parameter.
894 it should not be necessary to use this parameter.
882
895
883 **kw : optional
896 **kw : optional
884 keyword arguments that are passed down into the launcher
897 keyword arguments that are passed down into the launcher
885 callable.
898 callable.
886 """
899 """
887 if self.transport == 'tcp' and self.ip not in LOCAL_IPS:
900 if self.transport == 'tcp' and self.ip not in LOCAL_IPS:
888 raise RuntimeError("Can only launch a kernel on a local interface. "
901 raise RuntimeError("Can only launch a kernel on a local interface. "
889 "Make sure that the '*_address' attributes are "
902 "Make sure that the '*_address' attributes are "
890 "configured properly. "
903 "configured properly. "
891 "Currently valid addresses are: %s"%LOCAL_IPS
904 "Currently valid addresses are: %s"%LOCAL_IPS
892 )
905 )
893
906
894 # write connection file / get default ports
907 # write connection file / get default ports
895 self.write_connection_file()
908 self.write_connection_file()
896
909
897 self._launch_args = kw.copy()
910 self._launch_args = kw.copy()
898 launch_kernel = kw.pop('launcher', None)
911 launch_kernel = kw.pop('launcher', None)
899 if launch_kernel is None:
912 if launch_kernel is None:
900 from ipkernel import launch_kernel
913 from ipkernel import launch_kernel
901 self.kernel = launch_kernel(fname=self.connection_file, **kw)
914 self.kernel = launch_kernel(fname=self.connection_file, **kw)
902
915
903 def shutdown_kernel(self, now=False, restart=False):
916 def shutdown_kernel(self, now=False, restart=False):
904 """Attempts to the stop the kernel process cleanly.
917 """Attempts to the stop the kernel process cleanly.
905
918
906 This attempts to shutdown the kernels cleanly by:
919 This attempts to shutdown the kernels cleanly by:
907
920
908 1. Sending it a shutdown message over the shell channel.
921 1. Sending it a shutdown message over the shell channel.
909 2. If that fails, the kernel is shutdown forcibly by sending it
922 2. If that fails, the kernel is shutdown forcibly by sending it
910 a signal.
923 a signal.
911
924
912 Parameters:
925 Parameters:
913 -----------
926 -----------
914 now : bool
927 now : bool
915 Should the kernel be forcible killed *now*. This skips the
928 Should the kernel be forcible killed *now*. This skips the
916 first, nice shutdown attempt.
929 first, nice shutdown attempt.
917 restart: bool
930 restart: bool
918 Will this kernel be restarted after it is shutdown. When this
931 Will this kernel be restarted after it is shutdown. When this
919 is True, connection files will not be cleaned up.
932 is True, connection files will not be cleaned up.
920 """
933 """
921 # FIXME: Shutdown does not work on Windows due to ZMQ errors!
934 # FIXME: Shutdown does not work on Windows due to ZMQ errors!
922 if sys.platform == 'win32':
935 if sys.platform == 'win32':
923 self._kill_kernel()
936 self._kill_kernel()
924 return
937 return
925
938
926 # Pause the heart beat channel if it exists.
939 # Pause the heart beat channel if it exists.
927 if self._hb_channel is not None:
940 if self._hb_channel is not None:
928 self._hb_channel.pause()
941 self._hb_channel.pause()
929
942
930 if now:
943 if now:
931 if self.has_kernel:
944 if self.has_kernel:
932 self._kill_kernel()
945 self._kill_kernel()
933 else:
946 else:
934 # Don't send any additional kernel kill messages immediately, to give
947 # Don't send any additional kernel kill messages immediately, to give
935 # the kernel a chance to properly execute shutdown actions. Wait for at
948 # the kernel a chance to properly execute shutdown actions. Wait for at
936 # most 1s, checking every 0.1s.
949 # most 1s, checking every 0.1s.
937 self.shell_channel.shutdown(restart=restart)
950 self.shell_channel.shutdown(restart=restart)
938 for i in range(10):
951 for i in range(10):
939 if self.is_alive:
952 if self.is_alive:
940 time.sleep(0.1)
953 time.sleep(0.1)
941 else:
954 else:
942 break
955 break
943 else:
956 else:
944 # OK, we've waited long enough.
957 # OK, we've waited long enough.
945 if self.has_kernel:
958 if self.has_kernel:
946 self._kill_kernel()
959 self._kill_kernel()
947
960
948 if not restart:
961 if not restart:
949 self.cleanup_connection_file()
962 self.cleanup_connection_file()
950 self.cleanup_ipc_files()
963 self.cleanup_ipc_files()
951 else:
964 else:
952 self.cleanup_ipc_files()
965 self.cleanup_ipc_files()
953
966
954 def restart_kernel(self, now=False, **kw):
967 def restart_kernel(self, now=False, **kw):
955 """Restarts a kernel with the arguments that were used to launch it.
968 """Restarts a kernel with the arguments that were used to launch it.
956
969
957 If the old kernel was launched with random ports, the same ports will be
970 If the old kernel was launched with random ports, the same ports will be
958 used for the new kernel. The same connection file is used again.
971 used for the new kernel. The same connection file is used again.
959
972
960 Parameters
973 Parameters
961 ----------
974 ----------
962 now : bool, optional
975 now : bool, optional
963 If True, the kernel is forcefully restarted *immediately*, without
976 If True, the kernel is forcefully restarted *immediately*, without
964 having a chance to do any cleanup action. Otherwise the kernel is
977 having a chance to do any cleanup action. Otherwise the kernel is
965 given 1s to clean up before a forceful restart is issued.
978 given 1s to clean up before a forceful restart is issued.
966
979
967 In all cases the kernel is restarted, the only difference is whether
980 In all cases the kernel is restarted, the only difference is whether
968 it is given a chance to perform a clean shutdown or not.
981 it is given a chance to perform a clean shutdown or not.
969
982
970 **kw : optional
983 **kw : optional
971 Any options specified here will overwrite those used to launch the
984 Any options specified here will overwrite those used to launch the
972 kernel.
985 kernel.
973 """
986 """
974 if self._launch_args is None:
987 if self._launch_args is None:
975 raise RuntimeError("Cannot restart the kernel. "
988 raise RuntimeError("Cannot restart the kernel. "
976 "No previous call to 'start_kernel'.")
989 "No previous call to 'start_kernel'.")
977 else:
990 else:
978 # Stop currently running kernel.
991 # Stop currently running kernel.
979 self.shutdown_kernel(now=now, restart=True)
992 self.shutdown_kernel(now=now, restart=True)
980
993
981 # Start new kernel.
994 # Start new kernel.
982 self._launch_args.update(kw)
995 self._launch_args.update(kw)
983 self.start_kernel(**self._launch_args)
996 self.start_kernel(**self._launch_args)
984
997
985 # FIXME: Messages get dropped in Windows due to probable ZMQ bug
998 # FIXME: Messages get dropped in Windows due to probable ZMQ bug
986 # unless there is some delay here.
999 # unless there is some delay here.
987 if sys.platform == 'win32':
1000 if sys.platform == 'win32':
988 time.sleep(0.2)
1001 time.sleep(0.2)
989
1002
990 @property
1003 @property
991 def has_kernel(self):
1004 def has_kernel(self):
992 """Has a kernel been started that we are managing."""
1005 """Has a kernel been started that we are managing."""
993 return self.kernel is not None
1006 return self.kernel is not None
994
1007
995 def _kill_kernel(self):
1008 def _kill_kernel(self):
996 """Kill the running kernel.
1009 """Kill the running kernel.
997
1010
998 This is a private method, callers should use shutdown_kernel(now=True).
1011 This is a private method, callers should use shutdown_kernel(now=True).
999 """
1012 """
1000 if self.has_kernel:
1013 if self.has_kernel:
1001 # Pause the heart beat channel if it exists.
1014 # Pause the heart beat channel if it exists.
1002 if self._hb_channel is not None:
1015 if self._hb_channel is not None:
1003 self._hb_channel.pause()
1016 self._hb_channel.pause()
1004
1017
1005 # Signal the kernel to terminate (sends SIGKILL on Unix and calls
1018 # Signal the kernel to terminate (sends SIGKILL on Unix and calls
1006 # TerminateProcess() on Win32).
1019 # TerminateProcess() on Win32).
1007 try:
1020 try:
1008 self.kernel.kill()
1021 self.kernel.kill()
1009 except OSError as e:
1022 except OSError as e:
1010 # In Windows, we will get an Access Denied error if the process
1023 # In Windows, we will get an Access Denied error if the process
1011 # has already terminated. Ignore it.
1024 # has already terminated. Ignore it.
1012 if sys.platform == 'win32':
1025 if sys.platform == 'win32':
1013 if e.winerror != 5:
1026 if e.winerror != 5:
1014 raise
1027 raise
1015 # On Unix, we may get an ESRCH error if the process has already
1028 # On Unix, we may get an ESRCH error if the process has already
1016 # terminated. Ignore it.
1029 # terminated. Ignore it.
1017 else:
1030 else:
1018 from errno import ESRCH
1031 from errno import ESRCH
1019 if e.errno != ESRCH:
1032 if e.errno != ESRCH:
1020 raise
1033 raise
1021
1034
1022 # Block until the kernel terminates.
1035 # Block until the kernel terminates.
1023 self.kernel.wait()
1036 self.kernel.wait()
1024 self.kernel = None
1037 self.kernel = None
1025 else:
1038 else:
1026 raise RuntimeError("Cannot kill kernel. No kernel is running!")
1039 raise RuntimeError("Cannot kill kernel. No kernel is running!")
1027
1040
1028 def interrupt_kernel(self):
1041 def interrupt_kernel(self):
1029 """Interrupts the kernel by sending it a signal.
1042 """Interrupts the kernel by sending it a signal.
1030
1043
1031 Unlike ``signal_kernel``, this operation is well supported on all
1044 Unlike ``signal_kernel``, this operation is well supported on all
1032 platforms.
1045 platforms.
1033 """
1046 """
1034 if self.has_kernel:
1047 if self.has_kernel:
1035 if sys.platform == 'win32':
1048 if sys.platform == 'win32':
1036 from parentpoller import ParentPollerWindows as Poller
1049 from parentpoller import ParentPollerWindows as Poller
1037 Poller.send_interrupt(self.kernel.win32_interrupt_event)
1050 Poller.send_interrupt(self.kernel.win32_interrupt_event)
1038 else:
1051 else:
1039 self.kernel.send_signal(signal.SIGINT)
1052 self.kernel.send_signal(signal.SIGINT)
1040 else:
1053 else:
1041 raise RuntimeError("Cannot interrupt kernel. No kernel is running!")
1054 raise RuntimeError("Cannot interrupt kernel. No kernel is running!")
1042
1055
1043 def signal_kernel(self, signum):
1056 def signal_kernel(self, signum):
1044 """Sends a signal to the kernel.
1057 """Sends a signal to the kernel.
1045
1058
1046 Note that since only SIGTERM is supported on Windows, this function is
1059 Note that since only SIGTERM is supported on Windows, this function is
1047 only useful on Unix systems.
1060 only useful on Unix systems.
1048 """
1061 """
1049 if self.has_kernel:
1062 if self.has_kernel:
1050 self.kernel.send_signal(signum)
1063 self.kernel.send_signal(signum)
1051 else:
1064 else:
1052 raise RuntimeError("Cannot signal kernel. No kernel is running!")
1065 raise RuntimeError("Cannot signal kernel. No kernel is running!")
1053
1066
1054 @property
1067 @property
1055 def is_alive(self):
1068 def is_alive(self):
1056 """Is the kernel process still running?"""
1069 """Is the kernel process still running?"""
1057 if self.has_kernel:
1070 if self.has_kernel:
1058 if self.kernel.poll() is None:
1071 if self.kernel.poll() is None:
1059 return True
1072 return True
1060 else:
1073 else:
1061 return False
1074 return False
1062 elif self._hb_channel is not None:
1075 elif self._hb_channel is not None:
1063 # We didn't start the kernel with this KernelManager so we
1076 # We didn't start the kernel with this KernelManager so we
1064 # use the heartbeat.
1077 # use the heartbeat.
1065 return self._hb_channel.is_beating()
1078 return self._hb_channel.is_beating()
1066 else:
1079 else:
1067 # no heartbeat and not local, we can't tell if it's running,
1080 # no heartbeat and not local, we can't tell if it's running,
1068 # so naively return True
1081 # so naively return True
1069 return True
1082 return True
1070
1083
1071
1084
1072 #-----------------------------------------------------------------------------
1085 #-----------------------------------------------------------------------------
1073 # ABC Registration
1086 # ABC Registration
1074 #-----------------------------------------------------------------------------
1087 #-----------------------------------------------------------------------------
1075
1088
1076 ShellChannelABC.register(ShellChannel)
1089 ShellChannelABC.register(ShellChannel)
1077 IOPubChannelABC.register(IOPubChannel)
1090 IOPubChannelABC.register(IOPubChannel)
1078 HBChannelABC.register(HBChannel)
1091 HBChannelABC.register(HBChannel)
1079 StdInChannelABC.register(StdInChannel)
1092 StdInChannelABC.register(StdInChannel)
1080 KernelManagerABC.register(KernelManager)
1093 KernelManagerABC.register(KernelManager)
1081
1094
General Comments 0
You need to be logged in to leave comments. Login now