##// END OF EJS Templates
Commenting out atexit.register(context.close)
Brian Granger -
Show More
@@ -1,904 +1,905 b''
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-2010 The IPython Development Team
8 # Copyright (C) 2008-2010 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 from Queue import Queue, Empty
20 from Queue import Queue, Empty
21 from subprocess import Popen
21 from subprocess import Popen
22 import signal
22 import signal
23 import sys
23 import sys
24 from threading import Thread
24 from threading import Thread
25 import time
25 import time
26
26
27 # System library imports.
27 # System library imports.
28 import zmq
28 import zmq
29 from zmq import POLLIN, POLLOUT, POLLERR
29 from zmq import POLLIN, POLLOUT, POLLERR
30 from zmq.eventloop import ioloop
30 from zmq.eventloop import ioloop
31
31
32 # Local imports.
32 # Local imports.
33 from IPython.utils import io
33 from IPython.utils import io
34 from IPython.utils.traitlets import HasTraits, Any, Instance, Type, TCPAddress
34 from IPython.utils.traitlets import HasTraits, Any, Instance, Type, TCPAddress
35 from session import Session
35 from session import Session
36
36
37 #-----------------------------------------------------------------------------
37 #-----------------------------------------------------------------------------
38 # Constants and exceptions
38 # Constants and exceptions
39 #-----------------------------------------------------------------------------
39 #-----------------------------------------------------------------------------
40
40
41 LOCALHOST = '127.0.0.1'
41 LOCALHOST = '127.0.0.1'
42
42
43 class InvalidPortNumber(Exception):
43 class InvalidPortNumber(Exception):
44 pass
44 pass
45
45
46 #-----------------------------------------------------------------------------
46 #-----------------------------------------------------------------------------
47 # Utility functions
47 # Utility functions
48 #-----------------------------------------------------------------------------
48 #-----------------------------------------------------------------------------
49
49
50 # some utilities to validate message structure, these might get moved elsewhere
50 # some utilities to validate message structure, these might get moved elsewhere
51 # if they prove to have more generic utility
51 # if they prove to have more generic utility
52
52
53 def validate_string_list(lst):
53 def validate_string_list(lst):
54 """Validate that the input is a list of strings.
54 """Validate that the input is a list of strings.
55
55
56 Raises ValueError if not."""
56 Raises ValueError if not."""
57 if not isinstance(lst, list):
57 if not isinstance(lst, list):
58 raise ValueError('input %r must be a list' % lst)
58 raise ValueError('input %r must be a list' % lst)
59 for x in lst:
59 for x in lst:
60 if not isinstance(x, basestring):
60 if not isinstance(x, basestring):
61 raise ValueError('element %r in list must be a string' % x)
61 raise ValueError('element %r in list must be a string' % x)
62
62
63
63
64 def validate_string_dict(dct):
64 def validate_string_dict(dct):
65 """Validate that the input is a dict with string keys and values.
65 """Validate that the input is a dict with string keys and values.
66
66
67 Raises ValueError if not."""
67 Raises ValueError if not."""
68 for k,v in dct.iteritems():
68 for k,v in dct.iteritems():
69 if not isinstance(k, basestring):
69 if not isinstance(k, basestring):
70 raise ValueError('key %r in dict must be a string' % k)
70 raise ValueError('key %r in dict must be a string' % k)
71 if not isinstance(v, basestring):
71 if not isinstance(v, basestring):
72 raise ValueError('value %r in dict must be a string' % v)
72 raise ValueError('value %r in dict must be a string' % v)
73
73
74
74
75 #-----------------------------------------------------------------------------
75 #-----------------------------------------------------------------------------
76 # ZMQ Socket Channel classes
76 # ZMQ Socket Channel classes
77 #-----------------------------------------------------------------------------
77 #-----------------------------------------------------------------------------
78
78
79 class ZmqSocketChannel(Thread):
79 class ZmqSocketChannel(Thread):
80 """The base class for the channels that use ZMQ sockets.
80 """The base class for the channels that use ZMQ sockets.
81 """
81 """
82 context = None
82 context = None
83 session = None
83 session = None
84 socket = None
84 socket = None
85 ioloop = None
85 ioloop = None
86 iostate = None
86 iostate = None
87 _address = None
87 _address = None
88
88
89 def __init__(self, context, session, address):
89 def __init__(self, context, session, address):
90 """Create a channel
90 """Create a channel
91
91
92 Parameters
92 Parameters
93 ----------
93 ----------
94 context : :class:`zmq.Context`
94 context : :class:`zmq.Context`
95 The ZMQ context to use.
95 The ZMQ context to use.
96 session : :class:`session.Session`
96 session : :class:`session.Session`
97 The session to use.
97 The session to use.
98 address : tuple
98 address : tuple
99 Standard (ip, port) tuple that the kernel is listening on.
99 Standard (ip, port) tuple that the kernel is listening on.
100 """
100 """
101 super(ZmqSocketChannel, self).__init__()
101 super(ZmqSocketChannel, self).__init__()
102 self.daemon = True
102 self.daemon = True
103
103
104 self.context = context
104 self.context = context
105 self.session = session
105 self.session = session
106 if address[1] == 0:
106 if address[1] == 0:
107 message = 'The port number for a channel cannot be 0.'
107 message = 'The port number for a channel cannot be 0.'
108 raise InvalidPortNumber(message)
108 raise InvalidPortNumber(message)
109 self._address = address
109 self._address = address
110
110
111 def stop(self):
111 def stop(self):
112 """Stop the channel's activity.
112 """Stop the channel's activity.
113
113
114 This calls :method:`Thread.join` and returns when the thread
114 This calls :method:`Thread.join` and returns when the thread
115 terminates. :class:`RuntimeError` will be raised if
115 terminates. :class:`RuntimeError` will be raised if
116 :method:`self.start` is called again.
116 :method:`self.start` is called again.
117 """
117 """
118 self.join()
118 self.join()
119
119
120 @property
120 @property
121 def address(self):
121 def address(self):
122 """Get the channel's address as an (ip, port) tuple.
122 """Get the channel's address as an (ip, port) tuple.
123
123
124 By the default, the address is (localhost, 0), where 0 means a random
124 By the default, the address is (localhost, 0), where 0 means a random
125 port.
125 port.
126 """
126 """
127 return self._address
127 return self._address
128
128
129 def add_io_state(self, state):
129 def add_io_state(self, state):
130 """Add IO state to the eventloop.
130 """Add IO state to the eventloop.
131
131
132 Parameters
132 Parameters
133 ----------
133 ----------
134 state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
134 state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
135 The IO state flag to set.
135 The IO state flag to set.
136
136
137 This is thread safe as it uses the thread safe IOLoop.add_callback.
137 This is thread safe as it uses the thread safe IOLoop.add_callback.
138 """
138 """
139 def add_io_state_callback():
139 def add_io_state_callback():
140 if not self.iostate & state:
140 if not self.iostate & state:
141 self.iostate = self.iostate | state
141 self.iostate = self.iostate | state
142 self.ioloop.update_handler(self.socket, self.iostate)
142 self.ioloop.update_handler(self.socket, self.iostate)
143 self.ioloop.add_callback(add_io_state_callback)
143 self.ioloop.add_callback(add_io_state_callback)
144
144
145 def drop_io_state(self, state):
145 def drop_io_state(self, state):
146 """Drop IO state from the eventloop.
146 """Drop IO state from the eventloop.
147
147
148 Parameters
148 Parameters
149 ----------
149 ----------
150 state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
150 state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
151 The IO state flag to set.
151 The IO state flag to set.
152
152
153 This is thread safe as it uses the thread safe IOLoop.add_callback.
153 This is thread safe as it uses the thread safe IOLoop.add_callback.
154 """
154 """
155 def drop_io_state_callback():
155 def drop_io_state_callback():
156 if self.iostate & state:
156 if self.iostate & state:
157 self.iostate = self.iostate & (~state)
157 self.iostate = self.iostate & (~state)
158 self.ioloop.update_handler(self.socket, self.iostate)
158 self.ioloop.update_handler(self.socket, self.iostate)
159 self.ioloop.add_callback(drop_io_state_callback)
159 self.ioloop.add_callback(drop_io_state_callback)
160
160
161
161
162 class XReqSocketChannel(ZmqSocketChannel):
162 class XReqSocketChannel(ZmqSocketChannel):
163 """The XREQ channel for issues request/replies to the kernel.
163 """The XREQ channel for issues request/replies to the kernel.
164 """
164 """
165
165
166 command_queue = None
166 command_queue = None
167
167
168 def __init__(self, context, session, address):
168 def __init__(self, context, session, address):
169 super(XReqSocketChannel, self).__init__(context, session, address)
169 super(XReqSocketChannel, self).__init__(context, session, address)
170 self.command_queue = Queue()
170 self.command_queue = Queue()
171 self.ioloop = ioloop.IOLoop()
171 self.ioloop = ioloop.IOLoop()
172
172
173 def run(self):
173 def run(self):
174 """The thread's main activity. Call start() instead."""
174 """The thread's main activity. Call start() instead."""
175 self.socket = self.context.socket(zmq.XREQ)
175 self.socket = self.context.socket(zmq.XREQ)
176 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
176 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
177 self.socket.connect('tcp://%s:%i' % self.address)
177 self.socket.connect('tcp://%s:%i' % self.address)
178 self.iostate = POLLERR|POLLIN
178 self.iostate = POLLERR|POLLIN
179 self.ioloop.add_handler(self.socket, self._handle_events,
179 self.ioloop.add_handler(self.socket, self._handle_events,
180 self.iostate)
180 self.iostate)
181 self.ioloop.start()
181 self.ioloop.start()
182
182
183 def stop(self):
183 def stop(self):
184 self.ioloop.stop()
184 self.ioloop.stop()
185 super(XReqSocketChannel, self).stop()
185 super(XReqSocketChannel, self).stop()
186
186
187 def call_handlers(self, msg):
187 def call_handlers(self, msg):
188 """This method is called in the ioloop thread when a message arrives.
188 """This method is called in the ioloop thread when a message arrives.
189
189
190 Subclasses should override this method to handle incoming messages.
190 Subclasses should override this method to handle incoming messages.
191 It is important to remember that this method is called in the thread
191 It is important to remember that this method is called in the thread
192 so that some logic must be done to ensure that the application leve
192 so that some logic must be done to ensure that the application leve
193 handlers are called in the application thread.
193 handlers are called in the application thread.
194 """
194 """
195 raise NotImplementedError('call_handlers must be defined in a subclass.')
195 raise NotImplementedError('call_handlers must be defined in a subclass.')
196
196
197 def execute(self, code, silent=False,
197 def execute(self, code, silent=False,
198 user_variables=None, user_expressions=None):
198 user_variables=None, user_expressions=None):
199 """Execute code in the kernel.
199 """Execute code in the kernel.
200
200
201 Parameters
201 Parameters
202 ----------
202 ----------
203 code : str
203 code : str
204 A string of Python code.
204 A string of Python code.
205
205
206 silent : bool, optional (default False)
206 silent : bool, optional (default False)
207 If set, the kernel will execute the code as quietly possible.
207 If set, the kernel will execute the code as quietly possible.
208
208
209 user_variables : list, optional
209 user_variables : list, optional
210 A list of variable names to pull from the user's namespace. They
210 A list of variable names to pull from the user's namespace. They
211 will come back as a dict with these names as keys and their
211 will come back as a dict with these names as keys and their
212 :func:`repr` as values.
212 :func:`repr` as values.
213
213
214 user_expressions : dict, optional
214 user_expressions : dict, optional
215 A dict with string keys and to pull from the user's
215 A dict with string keys and to pull from the user's
216 namespace. They will come back as a dict with these names as keys
216 namespace. They will come back as a dict with these names as keys
217 and their :func:`repr` as values.
217 and their :func:`repr` as values.
218
218
219 Returns
219 Returns
220 -------
220 -------
221 The msg_id of the message sent.
221 The msg_id of the message sent.
222 """
222 """
223 if user_variables is None:
223 if user_variables is None:
224 user_variables = []
224 user_variables = []
225 if user_expressions is None:
225 if user_expressions is None:
226 user_expressions = {}
226 user_expressions = {}
227
227
228 # Don't waste network traffic if inputs are invalid
228 # Don't waste network traffic if inputs are invalid
229 if not isinstance(code, basestring):
229 if not isinstance(code, basestring):
230 raise ValueError('code %r must be a string' % code)
230 raise ValueError('code %r must be a string' % code)
231 validate_string_list(user_variables)
231 validate_string_list(user_variables)
232 validate_string_dict(user_expressions)
232 validate_string_dict(user_expressions)
233
233
234 # Create class for content/msg creation. Related to, but possibly
234 # Create class for content/msg creation. Related to, but possibly
235 # not in Session.
235 # not in Session.
236 content = dict(code=code, silent=silent,
236 content = dict(code=code, silent=silent,
237 user_variables=user_variables,
237 user_variables=user_variables,
238 user_expressions=user_expressions)
238 user_expressions=user_expressions)
239 msg = self.session.msg('execute_request', content)
239 msg = self.session.msg('execute_request', content)
240 self._queue_request(msg)
240 self._queue_request(msg)
241 return msg['header']['msg_id']
241 return msg['header']['msg_id']
242
242
243 def complete(self, text, line, cursor_pos, block=None):
243 def complete(self, text, line, cursor_pos, block=None):
244 """Tab complete text in the kernel's namespace.
244 """Tab complete text in the kernel's namespace.
245
245
246 Parameters
246 Parameters
247 ----------
247 ----------
248 text : str
248 text : str
249 The text to complete.
249 The text to complete.
250 line : str
250 line : str
251 The full line of text that is the surrounding context for the
251 The full line of text that is the surrounding context for the
252 text to complete.
252 text to complete.
253 cursor_pos : int
253 cursor_pos : int
254 The position of the cursor in the line where the completion was
254 The position of the cursor in the line where the completion was
255 requested.
255 requested.
256 block : str, optional
256 block : str, optional
257 The full block of code in which the completion is being requested.
257 The full block of code in which the completion is being requested.
258
258
259 Returns
259 Returns
260 -------
260 -------
261 The msg_id of the message sent.
261 The msg_id of the message sent.
262 """
262 """
263 content = dict(text=text, line=line, block=block, cursor_pos=cursor_pos)
263 content = dict(text=text, line=line, block=block, cursor_pos=cursor_pos)
264 msg = self.session.msg('complete_request', content)
264 msg = self.session.msg('complete_request', content)
265 self._queue_request(msg)
265 self._queue_request(msg)
266 return msg['header']['msg_id']
266 return msg['header']['msg_id']
267
267
268 def object_info(self, oname):
268 def object_info(self, oname):
269 """Get metadata information about an object.
269 """Get metadata information about an object.
270
270
271 Parameters
271 Parameters
272 ----------
272 ----------
273 oname : str
273 oname : str
274 A string specifying the object name.
274 A string specifying the object name.
275
275
276 Returns
276 Returns
277 -------
277 -------
278 The msg_id of the message sent.
278 The msg_id of the message sent.
279 """
279 """
280 content = dict(oname=oname)
280 content = dict(oname=oname)
281 msg = self.session.msg('object_info_request', content)
281 msg = self.session.msg('object_info_request', content)
282 self._queue_request(msg)
282 self._queue_request(msg)
283 return msg['header']['msg_id']
283 return msg['header']['msg_id']
284
284
285 def history(self, index=None, raw=False, output=True):
285 def history(self, index=None, raw=False, output=True):
286 """Get the history list.
286 """Get the history list.
287
287
288 Parameters
288 Parameters
289 ----------
289 ----------
290 index : n or (n1, n2) or None
290 index : n or (n1, n2) or None
291 If n, then the last entries. If a tuple, then all in
291 If n, then the last entries. If a tuple, then all in
292 range(n1, n2). If None, then all entries. Raises IndexError if
292 range(n1, n2). If None, then all entries. Raises IndexError if
293 the format of index is incorrect.
293 the format of index is incorrect.
294 raw : bool
294 raw : bool
295 If True, return the raw input.
295 If True, return the raw input.
296 output : bool
296 output : bool
297 If True, then return the output as well.
297 If True, then return the output as well.
298
298
299 Returns
299 Returns
300 -------
300 -------
301 The msg_id of the message sent.
301 The msg_id of the message sent.
302 """
302 """
303 content = dict(index=index, raw=raw, output=output)
303 content = dict(index=index, raw=raw, output=output)
304 msg = self.session.msg('history_request', content)
304 msg = self.session.msg('history_request', content)
305 self._queue_request(msg)
305 self._queue_request(msg)
306 return msg['header']['msg_id']
306 return msg['header']['msg_id']
307
307
308 def shutdown(self):
308 def shutdown(self):
309 """Request an immediate kernel shutdown.
309 """Request an immediate kernel shutdown.
310
310
311 Upon receipt of the (empty) reply, client code can safely assume that
311 Upon receipt of the (empty) reply, client code can safely assume that
312 the kernel has shut down and it's safe to forcefully terminate it if
312 the kernel has shut down and it's safe to forcefully terminate it if
313 it's still alive.
313 it's still alive.
314
314
315 The kernel will send the reply via a function registered with Python's
315 The kernel will send the reply via a function registered with Python's
316 atexit module, ensuring it's truly done as the kernel is done with all
316 atexit module, ensuring it's truly done as the kernel is done with all
317 normal operation.
317 normal operation.
318 """
318 """
319 # Send quit message to kernel. Once we implement kernel-side setattr,
319 # Send quit message to kernel. Once we implement kernel-side setattr,
320 # this should probably be done that way, but for now this will do.
320 # this should probably be done that way, but for now this will do.
321 msg = self.session.msg('shutdown_request', {})
321 msg = self.session.msg('shutdown_request', {})
322 self._queue_request(msg)
322 self._queue_request(msg)
323 return msg['header']['msg_id']
323 return msg['header']['msg_id']
324
324
325 def _handle_events(self, socket, events):
325 def _handle_events(self, socket, events):
326 if events & POLLERR:
326 if events & POLLERR:
327 self._handle_err()
327 self._handle_err()
328 if events & POLLOUT:
328 if events & POLLOUT:
329 self._handle_send()
329 self._handle_send()
330 if events & POLLIN:
330 if events & POLLIN:
331 self._handle_recv()
331 self._handle_recv()
332
332
333 def _handle_recv(self):
333 def _handle_recv(self):
334 msg = self.socket.recv_json()
334 msg = self.socket.recv_json()
335 self.call_handlers(msg)
335 self.call_handlers(msg)
336
336
337 def _handle_send(self):
337 def _handle_send(self):
338 try:
338 try:
339 msg = self.command_queue.get(False)
339 msg = self.command_queue.get(False)
340 except Empty:
340 except Empty:
341 pass
341 pass
342 else:
342 else:
343 self.socket.send_json(msg)
343 self.socket.send_json(msg)
344 if self.command_queue.empty():
344 if self.command_queue.empty():
345 self.drop_io_state(POLLOUT)
345 self.drop_io_state(POLLOUT)
346
346
347 def _handle_err(self):
347 def _handle_err(self):
348 # We don't want to let this go silently, so eventually we should log.
348 # We don't want to let this go silently, so eventually we should log.
349 raise zmq.ZMQError()
349 raise zmq.ZMQError()
350
350
351 def _queue_request(self, msg):
351 def _queue_request(self, msg):
352 self.command_queue.put(msg)
352 self.command_queue.put(msg)
353 self.add_io_state(POLLOUT)
353 self.add_io_state(POLLOUT)
354
354
355
355
356 class SubSocketChannel(ZmqSocketChannel):
356 class SubSocketChannel(ZmqSocketChannel):
357 """The SUB channel which listens for messages that the kernel publishes.
357 """The SUB channel which listens for messages that the kernel publishes.
358 """
358 """
359
359
360 def __init__(self, context, session, address):
360 def __init__(self, context, session, address):
361 super(SubSocketChannel, self).__init__(context, session, address)
361 super(SubSocketChannel, self).__init__(context, session, address)
362 self.ioloop = ioloop.IOLoop()
362 self.ioloop = ioloop.IOLoop()
363
363
364 def run(self):
364 def run(self):
365 """The thread's main activity. Call start() instead."""
365 """The thread's main activity. Call start() instead."""
366 self.socket = self.context.socket(zmq.SUB)
366 self.socket = self.context.socket(zmq.SUB)
367 self.socket.setsockopt(zmq.SUBSCRIBE,'')
367 self.socket.setsockopt(zmq.SUBSCRIBE,'')
368 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
368 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
369 self.socket.connect('tcp://%s:%i' % self.address)
369 self.socket.connect('tcp://%s:%i' % self.address)
370 self.iostate = POLLIN|POLLERR
370 self.iostate = POLLIN|POLLERR
371 self.ioloop.add_handler(self.socket, self._handle_events,
371 self.ioloop.add_handler(self.socket, self._handle_events,
372 self.iostate)
372 self.iostate)
373 self.ioloop.start()
373 self.ioloop.start()
374
374
375 def stop(self):
375 def stop(self):
376 self.ioloop.stop()
376 self.ioloop.stop()
377 super(SubSocketChannel, self).stop()
377 super(SubSocketChannel, self).stop()
378
378
379 def call_handlers(self, msg):
379 def call_handlers(self, msg):
380 """This method is called in the ioloop thread when a message arrives.
380 """This method is called in the ioloop thread when a message arrives.
381
381
382 Subclasses should override this method to handle incoming messages.
382 Subclasses should override this method to handle incoming messages.
383 It is important to remember that this method is called in the thread
383 It is important to remember that this method is called in the thread
384 so that some logic must be done to ensure that the application leve
384 so that some logic must be done to ensure that the application leve
385 handlers are called in the application thread.
385 handlers are called in the application thread.
386 """
386 """
387 raise NotImplementedError('call_handlers must be defined in a subclass.')
387 raise NotImplementedError('call_handlers must be defined in a subclass.')
388
388
389 def flush(self, timeout=1.0):
389 def flush(self, timeout=1.0):
390 """Immediately processes all pending messages on the SUB channel.
390 """Immediately processes all pending messages on the SUB channel.
391
391
392 Callers should use this method to ensure that :method:`call_handlers`
392 Callers should use this method to ensure that :method:`call_handlers`
393 has been called for all messages that have been received on the
393 has been called for all messages that have been received on the
394 0MQ SUB socket of this channel.
394 0MQ SUB socket of this channel.
395
395
396 This method is thread safe.
396 This method is thread safe.
397
397
398 Parameters
398 Parameters
399 ----------
399 ----------
400 timeout : float, optional
400 timeout : float, optional
401 The maximum amount of time to spend flushing, in seconds. The
401 The maximum amount of time to spend flushing, in seconds. The
402 default is one second.
402 default is one second.
403 """
403 """
404 # We do the IOLoop callback process twice to ensure that the IOLoop
404 # We do the IOLoop callback process twice to ensure that the IOLoop
405 # gets to perform at least one full poll.
405 # gets to perform at least one full poll.
406 stop_time = time.time() + timeout
406 stop_time = time.time() + timeout
407 for i in xrange(2):
407 for i in xrange(2):
408 self._flushed = False
408 self._flushed = False
409 self.ioloop.add_callback(self._flush)
409 self.ioloop.add_callback(self._flush)
410 while not self._flushed and time.time() < stop_time:
410 while not self._flushed and time.time() < stop_time:
411 time.sleep(0.01)
411 time.sleep(0.01)
412
412
413 def _handle_events(self, socket, events):
413 def _handle_events(self, socket, events):
414 # Turn on and off POLLOUT depending on if we have made a request
414 # Turn on and off POLLOUT depending on if we have made a request
415 if events & POLLERR:
415 if events & POLLERR:
416 self._handle_err()
416 self._handle_err()
417 if events & POLLIN:
417 if events & POLLIN:
418 self._handle_recv()
418 self._handle_recv()
419
419
420 def _handle_err(self):
420 def _handle_err(self):
421 # We don't want to let this go silently, so eventually we should log.
421 # We don't want to let this go silently, so eventually we should log.
422 raise zmq.ZMQError()
422 raise zmq.ZMQError()
423
423
424 def _handle_recv(self):
424 def _handle_recv(self):
425 # Get all of the messages we can
425 # Get all of the messages we can
426 while True:
426 while True:
427 try:
427 try:
428 msg = self.socket.recv_json(zmq.NOBLOCK)
428 msg = self.socket.recv_json(zmq.NOBLOCK)
429 except zmq.ZMQError:
429 except zmq.ZMQError:
430 # Check the errno?
430 # Check the errno?
431 # Will this trigger POLLERR?
431 # Will this trigger POLLERR?
432 break
432 break
433 else:
433 else:
434 self.call_handlers(msg)
434 self.call_handlers(msg)
435
435
436 def _flush(self):
436 def _flush(self):
437 """Callback for :method:`self.flush`."""
437 """Callback for :method:`self.flush`."""
438 self._flushed = True
438 self._flushed = True
439
439
440
440
441 class RepSocketChannel(ZmqSocketChannel):
441 class RepSocketChannel(ZmqSocketChannel):
442 """A reply channel to handle raw_input requests that the kernel makes."""
442 """A reply channel to handle raw_input requests that the kernel makes."""
443
443
444 msg_queue = None
444 msg_queue = None
445
445
446 def __init__(self, context, session, address):
446 def __init__(self, context, session, address):
447 super(RepSocketChannel, self).__init__(context, session, address)
447 super(RepSocketChannel, self).__init__(context, session, address)
448 self.ioloop = ioloop.IOLoop()
448 self.ioloop = ioloop.IOLoop()
449 self.msg_queue = Queue()
449 self.msg_queue = Queue()
450
450
451 def run(self):
451 def run(self):
452 """The thread's main activity. Call start() instead."""
452 """The thread's main activity. Call start() instead."""
453 self.socket = self.context.socket(zmq.XREQ)
453 self.socket = self.context.socket(zmq.XREQ)
454 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
454 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
455 self.socket.connect('tcp://%s:%i' % self.address)
455 self.socket.connect('tcp://%s:%i' % self.address)
456 self.iostate = POLLERR|POLLIN
456 self.iostate = POLLERR|POLLIN
457 self.ioloop.add_handler(self.socket, self._handle_events,
457 self.ioloop.add_handler(self.socket, self._handle_events,
458 self.iostate)
458 self.iostate)
459 self.ioloop.start()
459 self.ioloop.start()
460
460
461 def stop(self):
461 def stop(self):
462 self.ioloop.stop()
462 self.ioloop.stop()
463 super(RepSocketChannel, self).stop()
463 super(RepSocketChannel, self).stop()
464
464
465 def call_handlers(self, msg):
465 def call_handlers(self, msg):
466 """This method is called in the ioloop thread when a message arrives.
466 """This method is called in the ioloop thread when a message arrives.
467
467
468 Subclasses should override this method to handle incoming messages.
468 Subclasses should override this method to handle incoming messages.
469 It is important to remember that this method is called in the thread
469 It is important to remember that this method is called in the thread
470 so that some logic must be done to ensure that the application leve
470 so that some logic must be done to ensure that the application leve
471 handlers are called in the application thread.
471 handlers are called in the application thread.
472 """
472 """
473 raise NotImplementedError('call_handlers must be defined in a subclass.')
473 raise NotImplementedError('call_handlers must be defined in a subclass.')
474
474
475 def input(self, string):
475 def input(self, string):
476 """Send a string of raw input to the kernel."""
476 """Send a string of raw input to the kernel."""
477 content = dict(value=string)
477 content = dict(value=string)
478 msg = self.session.msg('input_reply', content)
478 msg = self.session.msg('input_reply', content)
479 self._queue_reply(msg)
479 self._queue_reply(msg)
480
480
481 def _handle_events(self, socket, events):
481 def _handle_events(self, socket, events):
482 if events & POLLERR:
482 if events & POLLERR:
483 self._handle_err()
483 self._handle_err()
484 if events & POLLOUT:
484 if events & POLLOUT:
485 self._handle_send()
485 self._handle_send()
486 if events & POLLIN:
486 if events & POLLIN:
487 self._handle_recv()
487 self._handle_recv()
488
488
489 def _handle_recv(self):
489 def _handle_recv(self):
490 msg = self.socket.recv_json()
490 msg = self.socket.recv_json()
491 self.call_handlers(msg)
491 self.call_handlers(msg)
492
492
493 def _handle_send(self):
493 def _handle_send(self):
494 try:
494 try:
495 msg = self.msg_queue.get(False)
495 msg = self.msg_queue.get(False)
496 except Empty:
496 except Empty:
497 pass
497 pass
498 else:
498 else:
499 self.socket.send_json(msg)
499 self.socket.send_json(msg)
500 if self.msg_queue.empty():
500 if self.msg_queue.empty():
501 self.drop_io_state(POLLOUT)
501 self.drop_io_state(POLLOUT)
502
502
503 def _handle_err(self):
503 def _handle_err(self):
504 # We don't want to let this go silently, so eventually we should log.
504 # We don't want to let this go silently, so eventually we should log.
505 raise zmq.ZMQError()
505 raise zmq.ZMQError()
506
506
507 def _queue_reply(self, msg):
507 def _queue_reply(self, msg):
508 self.msg_queue.put(msg)
508 self.msg_queue.put(msg)
509 self.add_io_state(POLLOUT)
509 self.add_io_state(POLLOUT)
510
510
511
511
512 class HBSocketChannel(ZmqSocketChannel):
512 class HBSocketChannel(ZmqSocketChannel):
513 """The heartbeat channel which monitors the kernel heartbeat.
513 """The heartbeat channel which monitors the kernel heartbeat.
514
514
515 Note that the heartbeat channel is paused by default. As long as you start
515 Note that the heartbeat channel is paused by default. As long as you start
516 this channel, the kernel manager will ensure that it is paused and un-paused
516 this channel, the kernel manager will ensure that it is paused and un-paused
517 as appropriate.
517 as appropriate.
518 """
518 """
519
519
520 time_to_dead = 3.0
520 time_to_dead = 3.0
521 socket = None
521 socket = None
522 poller = None
522 poller = None
523 _running = None
523 _running = None
524 _pause = None
524 _pause = None
525
525
526 def __init__(self, context, session, address):
526 def __init__(self, context, session, address):
527 super(HBSocketChannel, self).__init__(context, session, address)
527 super(HBSocketChannel, self).__init__(context, session, address)
528 self._running = False
528 self._running = False
529 self._pause = True
529 self._pause = True
530
530
531 def _create_socket(self):
531 def _create_socket(self):
532 self.socket = self.context.socket(zmq.REQ)
532 self.socket = self.context.socket(zmq.REQ)
533 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
533 self.socket.setsockopt(zmq.IDENTITY, self.session.session)
534 self.socket.connect('tcp://%s:%i' % self.address)
534 self.socket.connect('tcp://%s:%i' % self.address)
535 self.poller = zmq.Poller()
535 self.poller = zmq.Poller()
536 self.poller.register(self.socket, zmq.POLLIN)
536 self.poller.register(self.socket, zmq.POLLIN)
537
537
538 def run(self):
538 def run(self):
539 """The thread's main activity. Call start() instead."""
539 """The thread's main activity. Call start() instead."""
540 self._create_socket()
540 self._create_socket()
541 self._running = True
541 self._running = True
542 while self._running:
542 while self._running:
543 if self._pause:
543 if self._pause:
544 time.sleep(self.time_to_dead)
544 time.sleep(self.time_to_dead)
545 else:
545 else:
546 since_last_heartbeat = 0.0
546 since_last_heartbeat = 0.0
547 request_time = time.time()
547 request_time = time.time()
548 try:
548 try:
549 #io.rprint('Ping from HB channel') # dbg
549 #io.rprint('Ping from HB channel') # dbg
550 self.socket.send_json('ping')
550 self.socket.send_json('ping')
551 except zmq.ZMQError, e:
551 except zmq.ZMQError, e:
552 #io.rprint('*** HB Error:', e) # dbg
552 #io.rprint('*** HB Error:', e) # dbg
553 if e.errno == zmq.EFSM:
553 if e.errno == zmq.EFSM:
554 #io.rprint('sleep...', self.time_to_dead) # dbg
554 #io.rprint('sleep...', self.time_to_dead) # dbg
555 time.sleep(self.time_to_dead)
555 time.sleep(self.time_to_dead)
556 self._create_socket()
556 self._create_socket()
557 else:
557 else:
558 raise
558 raise
559 else:
559 else:
560 while True:
560 while True:
561 try:
561 try:
562 self.socket.recv_json(zmq.NOBLOCK)
562 self.socket.recv_json(zmq.NOBLOCK)
563 except zmq.ZMQError, e:
563 except zmq.ZMQError, e:
564 #io.rprint('*** HB Error 2:', e) # dbg
564 #io.rprint('*** HB Error 2:', e) # dbg
565 if e.errno == zmq.EAGAIN:
565 if e.errno == zmq.EAGAIN:
566 before_poll = time.time()
566 before_poll = time.time()
567 until_dead = self.time_to_dead - (before_poll -
567 until_dead = self.time_to_dead - (before_poll -
568 request_time)
568 request_time)
569
569
570 # When the return value of poll() is an empty
570 # When the return value of poll() is an empty
571 # list, that is when things have gone wrong
571 # list, that is when things have gone wrong
572 # (zeromq bug). As long as it is not an empty
572 # (zeromq bug). As long as it is not an empty
573 # list, poll is working correctly even if it
573 # list, poll is working correctly even if it
574 # returns quickly. Note: poll timeout is in
574 # returns quickly. Note: poll timeout is in
575 # milliseconds.
575 # milliseconds.
576 self.poller.poll(1000*until_dead)
576 self.poller.poll(1000*until_dead)
577
577
578 since_last_heartbeat = time.time()-request_time
578 since_last_heartbeat = time.time()-request_time
579 if since_last_heartbeat > self.time_to_dead:
579 if since_last_heartbeat > self.time_to_dead:
580 self.call_handlers(since_last_heartbeat)
580 self.call_handlers(since_last_heartbeat)
581 break
581 break
582 else:
582 else:
583 # FIXME: We should probably log this instead.
583 # FIXME: We should probably log this instead.
584 raise
584 raise
585 else:
585 else:
586 until_dead = self.time_to_dead - (time.time() -
586 until_dead = self.time_to_dead - (time.time() -
587 request_time)
587 request_time)
588 if until_dead > 0.0:
588 if until_dead > 0.0:
589 #io.rprint('sleep...', self.time_to_dead) # dbg
589 #io.rprint('sleep...', self.time_to_dead) # dbg
590 time.sleep(until_dead)
590 time.sleep(until_dead)
591 break
591 break
592
592
593 def pause(self):
593 def pause(self):
594 """Pause the heartbeat."""
594 """Pause the heartbeat."""
595 self._pause = True
595 self._pause = True
596
596
597 def unpause(self):
597 def unpause(self):
598 """Unpause the heartbeat."""
598 """Unpause the heartbeat."""
599 self._pause = False
599 self._pause = False
600
600
601 def is_beating(self):
601 def is_beating(self):
602 """Is the heartbeat running and not paused."""
602 """Is the heartbeat running and not paused."""
603 if self.is_alive() and not self._pause:
603 if self.is_alive() and not self._pause:
604 return True
604 return True
605 else:
605 else:
606 return False
606 return False
607
607
608 def stop(self):
608 def stop(self):
609 self._running = False
609 self._running = False
610 super(HBSocketChannel, self).stop()
610 super(HBSocketChannel, self).stop()
611
611
612 def call_handlers(self, since_last_heartbeat):
612 def call_handlers(self, since_last_heartbeat):
613 """This method is called in the ioloop thread when a message arrives.
613 """This method is called in the ioloop thread when a message arrives.
614
614
615 Subclasses should override this method to handle incoming messages.
615 Subclasses should override this method to handle incoming messages.
616 It is important to remember that this method is called in the thread
616 It is important to remember that this method is called in the thread
617 so that some logic must be done to ensure that the application leve
617 so that some logic must be done to ensure that the application leve
618 handlers are called in the application thread.
618 handlers are called in the application thread.
619 """
619 """
620 raise NotImplementedError('call_handlers must be defined in a subclass.')
620 raise NotImplementedError('call_handlers must be defined in a subclass.')
621
621
622
622
623 #-----------------------------------------------------------------------------
623 #-----------------------------------------------------------------------------
624 # Main kernel manager class
624 # Main kernel manager class
625 #-----------------------------------------------------------------------------
625 #-----------------------------------------------------------------------------
626
626
627 class KernelManager(HasTraits):
627 class KernelManager(HasTraits):
628 """ Manages a kernel for a frontend.
628 """ Manages a kernel for a frontend.
629
629
630 The SUB channel is for the frontend to receive messages published by the
630 The SUB channel is for the frontend to receive messages published by the
631 kernel.
631 kernel.
632
632
633 The REQ channel is for the frontend to make requests of the kernel.
633 The REQ channel is for the frontend to make requests of the kernel.
634
634
635 The REP channel is for the kernel to request stdin (raw_input) from the
635 The REP channel is for the kernel to request stdin (raw_input) from the
636 frontend.
636 frontend.
637 """
637 """
638 # The PyZMQ Context to use for communication with the kernel.
638 # The PyZMQ Context to use for communication with the kernel.
639 context = Instance(zmq.Context,(),{})
639 context = Instance(zmq.Context,(),{})
640
640
641 # The Session to use for communication with the kernel.
641 # The Session to use for communication with the kernel.
642 session = Instance(Session,(),{})
642 session = Instance(Session,(),{})
643
643
644 # The kernel process with which the KernelManager is communicating.
644 # The kernel process with which the KernelManager is communicating.
645 kernel = Instance(Popen)
645 kernel = Instance(Popen)
646
646
647 # The addresses for the communication channels.
647 # The addresses for the communication channels.
648 xreq_address = TCPAddress((LOCALHOST, 0))
648 xreq_address = TCPAddress((LOCALHOST, 0))
649 sub_address = TCPAddress((LOCALHOST, 0))
649 sub_address = TCPAddress((LOCALHOST, 0))
650 rep_address = TCPAddress((LOCALHOST, 0))
650 rep_address = TCPAddress((LOCALHOST, 0))
651 hb_address = TCPAddress((LOCALHOST, 0))
651 hb_address = TCPAddress((LOCALHOST, 0))
652
652
653 # The classes to use for the various channels.
653 # The classes to use for the various channels.
654 xreq_channel_class = Type(XReqSocketChannel)
654 xreq_channel_class = Type(XReqSocketChannel)
655 sub_channel_class = Type(SubSocketChannel)
655 sub_channel_class = Type(SubSocketChannel)
656 rep_channel_class = Type(RepSocketChannel)
656 rep_channel_class = Type(RepSocketChannel)
657 hb_channel_class = Type(HBSocketChannel)
657 hb_channel_class = Type(HBSocketChannel)
658
658
659 # Protected traits.
659 # Protected traits.
660 _launch_args = Any
660 _launch_args = Any
661 _xreq_channel = Any
661 _xreq_channel = Any
662 _sub_channel = Any
662 _sub_channel = Any
663 _rep_channel = Any
663 _rep_channel = Any
664 _hb_channel = Any
664 _hb_channel = Any
665
665
666 def __init__(self, **kwargs):
666 def __init__(self, **kwargs):
667 super(KernelManager, self).__init__(**kwargs)
667 super(KernelManager, self).__init__(**kwargs)
668 atexit.register(self.context.close)
668 # Uncomment this to try closing the context.
669 # atexit.register(self.context.close)
669
670
670 #--------------------------------------------------------------------------
671 #--------------------------------------------------------------------------
671 # Channel management methods:
672 # Channel management methods:
672 #--------------------------------------------------------------------------
673 #--------------------------------------------------------------------------
673
674
674 def start_channels(self, xreq=True, sub=True, rep=True, hb=True):
675 def start_channels(self, xreq=True, sub=True, rep=True, hb=True):
675 """Starts the channels for this kernel.
676 """Starts the channels for this kernel.
676
677
677 This will create the channels if they do not exist and then start
678 This will create the channels if they do not exist and then start
678 them. If port numbers of 0 are being used (random ports) then you
679 them. If port numbers of 0 are being used (random ports) then you
679 must first call :method:`start_kernel`. If the channels have been
680 must first call :method:`start_kernel`. If the channels have been
680 stopped and you call this, :class:`RuntimeError` will be raised.
681 stopped and you call this, :class:`RuntimeError` will be raised.
681 """
682 """
682 if xreq:
683 if xreq:
683 self.xreq_channel.start()
684 self.xreq_channel.start()
684 if sub:
685 if sub:
685 self.sub_channel.start()
686 self.sub_channel.start()
686 if rep:
687 if rep:
687 self.rep_channel.start()
688 self.rep_channel.start()
688 if hb:
689 if hb:
689 self.hb_channel.start()
690 self.hb_channel.start()
690
691
691 def stop_channels(self):
692 def stop_channels(self):
692 """Stops all the running channels for this kernel.
693 """Stops all the running channels for this kernel.
693 """
694 """
694 if self.xreq_channel.is_alive():
695 if self.xreq_channel.is_alive():
695 self.xreq_channel.stop()
696 self.xreq_channel.stop()
696 if self.sub_channel.is_alive():
697 if self.sub_channel.is_alive():
697 self.sub_channel.stop()
698 self.sub_channel.stop()
698 if self.rep_channel.is_alive():
699 if self.rep_channel.is_alive():
699 self.rep_channel.stop()
700 self.rep_channel.stop()
700 if self.hb_channel.is_alive():
701 if self.hb_channel.is_alive():
701 self.hb_channel.stop()
702 self.hb_channel.stop()
702
703
703 @property
704 @property
704 def channels_running(self):
705 def channels_running(self):
705 """Are any of the channels created and running?"""
706 """Are any of the channels created and running?"""
706 return (self.xreq_channel.is_alive() or self.sub_channel.is_alive() or
707 return (self.xreq_channel.is_alive() or self.sub_channel.is_alive() or
707 self.rep_channel.is_alive() or self.hb_channel.is_alive())
708 self.rep_channel.is_alive() or self.hb_channel.is_alive())
708
709
709 #--------------------------------------------------------------------------
710 #--------------------------------------------------------------------------
710 # Kernel process management methods:
711 # Kernel process management methods:
711 #--------------------------------------------------------------------------
712 #--------------------------------------------------------------------------
712
713
713 def start_kernel(self, **kw):
714 def start_kernel(self, **kw):
714 """Starts a kernel process and configures the manager to use it.
715 """Starts a kernel process and configures the manager to use it.
715
716
716 If random ports (port=0) are being used, this method must be called
717 If random ports (port=0) are being used, this method must be called
717 before the channels are created.
718 before the channels are created.
718
719
719 Parameters:
720 Parameters:
720 -----------
721 -----------
721 ipython : bool, optional (default True)
722 ipython : bool, optional (default True)
722 Whether to use an IPython kernel instead of a plain Python kernel.
723 Whether to use an IPython kernel instead of a plain Python kernel.
723 """
724 """
724 xreq, sub, rep, hb = self.xreq_address, self.sub_address, \
725 xreq, sub, rep, hb = self.xreq_address, self.sub_address, \
725 self.rep_address, self.hb_address
726 self.rep_address, self.hb_address
726 if xreq[0] != LOCALHOST or sub[0] != LOCALHOST or \
727 if xreq[0] != LOCALHOST or sub[0] != LOCALHOST or \
727 rep[0] != LOCALHOST or hb[0] != LOCALHOST:
728 rep[0] != LOCALHOST or hb[0] != LOCALHOST:
728 raise RuntimeError("Can only launch a kernel on localhost."
729 raise RuntimeError("Can only launch a kernel on localhost."
729 "Make sure that the '*_address' attributes are "
730 "Make sure that the '*_address' attributes are "
730 "configured properly.")
731 "configured properly.")
731
732
732 self._launch_args = kw.copy()
733 self._launch_args = kw.copy()
733 if kw.pop('ipython', True):
734 if kw.pop('ipython', True):
734 from ipkernel import launch_kernel
735 from ipkernel import launch_kernel
735 else:
736 else:
736 from pykernel import launch_kernel
737 from pykernel import launch_kernel
737 self.kernel, xrep, pub, req, hb = launch_kernel(
738 self.kernel, xrep, pub, req, hb = launch_kernel(
738 xrep_port=xreq[1], pub_port=sub[1],
739 xrep_port=xreq[1], pub_port=sub[1],
739 req_port=rep[1], hb_port=hb[1], **kw)
740 req_port=rep[1], hb_port=hb[1], **kw)
740 self.xreq_address = (LOCALHOST, xrep)
741 self.xreq_address = (LOCALHOST, xrep)
741 self.sub_address = (LOCALHOST, pub)
742 self.sub_address = (LOCALHOST, pub)
742 self.rep_address = (LOCALHOST, req)
743 self.rep_address = (LOCALHOST, req)
743 self.hb_address = (LOCALHOST, hb)
744 self.hb_address = (LOCALHOST, hb)
744
745
745 def shutdown_kernel(self):
746 def shutdown_kernel(self):
746 """ Attempts to the stop the kernel process cleanly. If the kernel
747 """ Attempts to the stop the kernel process cleanly. If the kernel
747 cannot be stopped, it is killed, if possible.
748 cannot be stopped, it is killed, if possible.
748 """
749 """
749 # FIXME: Shutdown does not work on Windows due to ZMQ errors!
750 # FIXME: Shutdown does not work on Windows due to ZMQ errors!
750 if sys.platform == 'win32':
751 if sys.platform == 'win32':
751 self.kill_kernel()
752 self.kill_kernel()
752 return
753 return
753
754
754 # Pause the heart beat channel if it exists.
755 # Pause the heart beat channel if it exists.
755 if self._hb_channel is not None:
756 if self._hb_channel is not None:
756 self._hb_channel.pause()
757 self._hb_channel.pause()
757
758
758 # Don't send any additional kernel kill messages immediately, to give
759 # Don't send any additional kernel kill messages immediately, to give
759 # the kernel a chance to properly execute shutdown actions. Wait for at
760 # the kernel a chance to properly execute shutdown actions. Wait for at
760 # most 1s, checking every 0.1s.
761 # most 1s, checking every 0.1s.
761 self.xreq_channel.shutdown()
762 self.xreq_channel.shutdown()
762 for i in range(10):
763 for i in range(10):
763 if self.is_alive:
764 if self.is_alive:
764 time.sleep(0.1)
765 time.sleep(0.1)
765 else:
766 else:
766 break
767 break
767 else:
768 else:
768 # OK, we've waited long enough.
769 # OK, we've waited long enough.
769 if self.has_kernel:
770 if self.has_kernel:
770 self.kill_kernel()
771 self.kill_kernel()
771
772
772 def restart_kernel(self, now=False):
773 def restart_kernel(self, now=False):
773 """Restarts a kernel with the same arguments that were used to launch
774 """Restarts a kernel with the same arguments that were used to launch
774 it. If the old kernel was launched with random ports, the same ports
775 it. If the old kernel was launched with random ports, the same ports
775 will be used for the new kernel.
776 will be used for the new kernel.
776
777
777 Parameters
778 Parameters
778 ----------
779 ----------
779 now : bool, optional
780 now : bool, optional
780 If True, the kernel is forcefully restarted *immediately*, without
781 If True, the kernel is forcefully restarted *immediately*, without
781 having a chance to do any cleanup action. Otherwise the kernel is
782 having a chance to do any cleanup action. Otherwise the kernel is
782 given 1s to clean up before a forceful restart is issued.
783 given 1s to clean up before a forceful restart is issued.
783
784
784 In all cases the kernel is restarted, the only difference is whether
785 In all cases the kernel is restarted, the only difference is whether
785 it is given a chance to perform a clean shutdown or not.
786 it is given a chance to perform a clean shutdown or not.
786 """
787 """
787 if self._launch_args is None:
788 if self._launch_args is None:
788 raise RuntimeError("Cannot restart the kernel. "
789 raise RuntimeError("Cannot restart the kernel. "
789 "No previous call to 'start_kernel'.")
790 "No previous call to 'start_kernel'.")
790 else:
791 else:
791 if self.has_kernel:
792 if self.has_kernel:
792 if now:
793 if now:
793 self.kill_kernel()
794 self.kill_kernel()
794 else:
795 else:
795 self.shutdown_kernel()
796 self.shutdown_kernel()
796 self.start_kernel(**self._launch_args)
797 self.start_kernel(**self._launch_args)
797
798
798 # FIXME: Messages get dropped in Windows due to probable ZMQ bug
799 # FIXME: Messages get dropped in Windows due to probable ZMQ bug
799 # unless there is some delay here.
800 # unless there is some delay here.
800 if sys.platform == 'win32':
801 if sys.platform == 'win32':
801 time.sleep(0.2)
802 time.sleep(0.2)
802
803
803 @property
804 @property
804 def has_kernel(self):
805 def has_kernel(self):
805 """Returns whether a kernel process has been specified for the kernel
806 """Returns whether a kernel process has been specified for the kernel
806 manager.
807 manager.
807 """
808 """
808 return self.kernel is not None
809 return self.kernel is not None
809
810
810 def kill_kernel(self):
811 def kill_kernel(self):
811 """ Kill the running kernel. """
812 """ Kill the running kernel. """
812 if self.has_kernel:
813 if self.has_kernel:
813 # Pause the heart beat channel if it exists.
814 # Pause the heart beat channel if it exists.
814 if self._hb_channel is not None:
815 if self._hb_channel is not None:
815 self._hb_channel.pause()
816 self._hb_channel.pause()
816
817
817 # Attempt to kill the kernel.
818 # Attempt to kill the kernel.
818 try:
819 try:
819 self.kernel.kill()
820 self.kernel.kill()
820 except OSError, e:
821 except OSError, e:
821 # In Windows, we will get an Access Denied error if the process
822 # In Windows, we will get an Access Denied error if the process
822 # has already terminated. Ignore it.
823 # has already terminated. Ignore it.
823 if not (sys.platform == 'win32' and e.winerror == 5):
824 if not (sys.platform == 'win32' and e.winerror == 5):
824 raise
825 raise
825 self.kernel = None
826 self.kernel = None
826 else:
827 else:
827 raise RuntimeError("Cannot kill kernel. No kernel is running!")
828 raise RuntimeError("Cannot kill kernel. No kernel is running!")
828
829
829 def interrupt_kernel(self):
830 def interrupt_kernel(self):
830 """ Interrupts the kernel. Unlike ``signal_kernel``, this operation is
831 """ Interrupts the kernel. Unlike ``signal_kernel``, this operation is
831 well supported on all platforms.
832 well supported on all platforms.
832 """
833 """
833 if self.has_kernel:
834 if self.has_kernel:
834 if sys.platform == 'win32':
835 if sys.platform == 'win32':
835 from parentpoller import ParentPollerWindows as Poller
836 from parentpoller import ParentPollerWindows as Poller
836 Poller.send_interrupt(self.kernel.win32_interrupt_event)
837 Poller.send_interrupt(self.kernel.win32_interrupt_event)
837 else:
838 else:
838 self.kernel.send_signal(signal.SIGINT)
839 self.kernel.send_signal(signal.SIGINT)
839 else:
840 else:
840 raise RuntimeError("Cannot interrupt kernel. No kernel is running!")
841 raise RuntimeError("Cannot interrupt kernel. No kernel is running!")
841
842
842 def signal_kernel(self, signum):
843 def signal_kernel(self, signum):
843 """ Sends a signal to the kernel. Note that since only SIGTERM is
844 """ Sends a signal to the kernel. Note that since only SIGTERM is
844 supported on Windows, this function is only useful on Unix systems.
845 supported on Windows, this function is only useful on Unix systems.
845 """
846 """
846 if self.has_kernel:
847 if self.has_kernel:
847 self.kernel.send_signal(signum)
848 self.kernel.send_signal(signum)
848 else:
849 else:
849 raise RuntimeError("Cannot signal kernel. No kernel is running!")
850 raise RuntimeError("Cannot signal kernel. No kernel is running!")
850
851
851 @property
852 @property
852 def is_alive(self):
853 def is_alive(self):
853 """Is the kernel process still running?"""
854 """Is the kernel process still running?"""
854 # FIXME: not using a heartbeat means this method is broken for any
855 # FIXME: not using a heartbeat means this method is broken for any
855 # remote kernel, it's only capable of handling local kernels.
856 # remote kernel, it's only capable of handling local kernels.
856 if self.has_kernel:
857 if self.has_kernel:
857 if self.kernel.poll() is None:
858 if self.kernel.poll() is None:
858 return True
859 return True
859 else:
860 else:
860 return False
861 return False
861 else:
862 else:
862 # We didn't start the kernel with this KernelManager so we don't
863 # We didn't start the kernel with this KernelManager so we don't
863 # know if it is running. We should use a heartbeat for this case.
864 # know if it is running. We should use a heartbeat for this case.
864 return True
865 return True
865
866
866 #--------------------------------------------------------------------------
867 #--------------------------------------------------------------------------
867 # Channels used for communication with the kernel:
868 # Channels used for communication with the kernel:
868 #--------------------------------------------------------------------------
869 #--------------------------------------------------------------------------
869
870
870 @property
871 @property
871 def xreq_channel(self):
872 def xreq_channel(self):
872 """Get the REQ socket channel object to make requests of the kernel."""
873 """Get the REQ socket channel object to make requests of the kernel."""
873 if self._xreq_channel is None:
874 if self._xreq_channel is None:
874 self._xreq_channel = self.xreq_channel_class(self.context,
875 self._xreq_channel = self.xreq_channel_class(self.context,
875 self.session,
876 self.session,
876 self.xreq_address)
877 self.xreq_address)
877 return self._xreq_channel
878 return self._xreq_channel
878
879
879 @property
880 @property
880 def sub_channel(self):
881 def sub_channel(self):
881 """Get the SUB socket channel object."""
882 """Get the SUB socket channel object."""
882 if self._sub_channel is None:
883 if self._sub_channel is None:
883 self._sub_channel = self.sub_channel_class(self.context,
884 self._sub_channel = self.sub_channel_class(self.context,
884 self.session,
885 self.session,
885 self.sub_address)
886 self.sub_address)
886 return self._sub_channel
887 return self._sub_channel
887
888
888 @property
889 @property
889 def rep_channel(self):
890 def rep_channel(self):
890 """Get the REP socket channel object to handle stdin (raw_input)."""
891 """Get the REP socket channel object to handle stdin (raw_input)."""
891 if self._rep_channel is None:
892 if self._rep_channel is None:
892 self._rep_channel = self.rep_channel_class(self.context,
893 self._rep_channel = self.rep_channel_class(self.context,
893 self.session,
894 self.session,
894 self.rep_address)
895 self.rep_address)
895 return self._rep_channel
896 return self._rep_channel
896
897
897 @property
898 @property
898 def hb_channel(self):
899 def hb_channel(self):
899 """Get the REP socket channel object to handle stdin (raw_input)."""
900 """Get the REP socket channel object to handle stdin (raw_input)."""
900 if self._hb_channel is None:
901 if self._hb_channel is None:
901 self._hb_channel = self.hb_channel_class(self.context,
902 self._hb_channel = self.hb_channel_class(self.context,
902 self.session,
903 self.session,
903 self.hb_address)
904 self.hb_address)
904 return self._hb_channel
905 return self._hb_channel
General Comments 0
You need to be logged in to leave comments. Login now