##// END OF EJS Templates
Remove redundant make_*_socket functions
Thomas Kluyver -
Show More
@@ -1,226 +1,203 b''
1 """Base classes to manage a Client's interaction with a running kernel"""
1 """Base classes to manage a Client's interaction with a running kernel"""
2
2
3 # Copyright (c) IPython Development Team.
3 # Copyright (c) IPython Development Team.
4 # Distributed under the terms of the Modified BSD License.
4 # Distributed under the terms of the Modified BSD License.
5
5
6 from __future__ import absolute_import
6 from __future__ import absolute_import
7
7
8 import atexit
8 import atexit
9 import errno
9 import errno
10 from threading import Thread
10 from threading import Thread
11 import time
11 import time
12
12
13 import zmq
13 import zmq
14 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
14 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
15 # during garbage collection of threads at exit:
15 # during garbage collection of threads at exit:
16 from zmq import ZMQError
16 from zmq import ZMQError
17
17
18 from IPython.core.release import kernel_protocol_version_info
18 from IPython.core.release import kernel_protocol_version_info
19
19
20 from .channelsabc import HBChannelABC
20 from .channelsabc import HBChannelABC
21
21
22 #-----------------------------------------------------------------------------
22 #-----------------------------------------------------------------------------
23 # Constants and exceptions
23 # Constants and exceptions
24 #-----------------------------------------------------------------------------
24 #-----------------------------------------------------------------------------
25
25
26 major_protocol_version = kernel_protocol_version_info[0]
26 major_protocol_version = kernel_protocol_version_info[0]
27
27
28 class InvalidPortNumber(Exception):
28 class InvalidPortNumber(Exception):
29 pass
29 pass
30
30
31
32 def make_shell_socket(context, identity, address):
33 socket = context.socket(zmq.DEALER)
34 socket.linger = 1000
35 socket.identity = identity
36 socket.connect(address)
37 return socket
38
39 def make_iopub_socket(context, identity, address):
40 socket = context.socket(zmq.SUB)
41 socket.linger = 1000
42 socket.subscribe = b''
43 socket.identity = identity
44 socket.connect(address)
45 return socket
46
47 def make_stdin_socket(context, identity, address):
48 socket = context.socket(zmq.DEALER)
49 socket.linger = 1000
50 socket.identity = identity
51 socket.connect(address)
52 return socket
53
54 class HBChannel(Thread):
31 class HBChannel(Thread):
55 """The heartbeat channel which monitors the kernel heartbeat.
32 """The heartbeat channel which monitors the kernel heartbeat.
56
33
57 Note that the heartbeat channel is paused by default. As long as you start
34 Note that the heartbeat channel is paused by default. As long as you start
58 this channel, the kernel manager will ensure that it is paused and un-paused
35 this channel, the kernel manager will ensure that it is paused and un-paused
59 as appropriate.
36 as appropriate.
60 """
37 """
61 context = None
38 context = None
62 session = None
39 session = None
63 socket = None
40 socket = None
64 address = None
41 address = None
65 _exiting = False
42 _exiting = False
66
43
67 time_to_dead = 1.
44 time_to_dead = 1.
68 poller = None
45 poller = None
69 _running = None
46 _running = None
70 _pause = None
47 _pause = None
71 _beating = None
48 _beating = None
72
49
73 def __init__(self, context, session, address):
50 def __init__(self, context, session, address):
74 """Create the heartbeat monitor thread.
51 """Create the heartbeat monitor thread.
75
52
76 Parameters
53 Parameters
77 ----------
54 ----------
78 context : :class:`zmq.Context`
55 context : :class:`zmq.Context`
79 The ZMQ context to use.
56 The ZMQ context to use.
80 session : :class:`session.Session`
57 session : :class:`session.Session`
81 The session to use.
58 The session to use.
82 address : zmq url
59 address : zmq url
83 Standard (ip, port) tuple that the kernel is listening on.
60 Standard (ip, port) tuple that the kernel is listening on.
84 """
61 """
85 super(HBChannel, self).__init__()
62 super(HBChannel, self).__init__()
86 self.daemon = True
63 self.daemon = True
87
64
88 self.context = context
65 self.context = context
89 self.session = session
66 self.session = session
90 if isinstance(address, tuple):
67 if isinstance(address, tuple):
91 if address[1] == 0:
68 if address[1] == 0:
92 message = 'The port number for a channel cannot be 0.'
69 message = 'The port number for a channel cannot be 0.'
93 raise InvalidPortNumber(message)
70 raise InvalidPortNumber(message)
94 address = "tcp://%s:%i" % address
71 address = "tcp://%s:%i" % address
95 self.address = address
72 self.address = address
96 atexit.register(self._notice_exit)
73 atexit.register(self._notice_exit)
97
74
98 self._running = False
75 self._running = False
99 self._pause = True
76 self._pause = True
100 self.poller = zmq.Poller()
77 self.poller = zmq.Poller()
101
78
102 def _notice_exit(self):
79 def _notice_exit(self):
103 self._exiting = True
80 self._exiting = True
104
81
105 def _create_socket(self):
82 def _create_socket(self):
106 if self.socket is not None:
83 if self.socket is not None:
107 # close previous socket, before opening a new one
84 # close previous socket, before opening a new one
108 self.poller.unregister(self.socket)
85 self.poller.unregister(self.socket)
109 self.socket.close()
86 self.socket.close()
110 self.socket = self.context.socket(zmq.REQ)
87 self.socket = self.context.socket(zmq.REQ)
111 self.socket.linger = 1000
88 self.socket.linger = 1000
112 self.socket.connect(self.address)
89 self.socket.connect(self.address)
113
90
114 self.poller.register(self.socket, zmq.POLLIN)
91 self.poller.register(self.socket, zmq.POLLIN)
115
92
116 def _poll(self, start_time):
93 def _poll(self, start_time):
117 """poll for heartbeat replies until we reach self.time_to_dead.
94 """poll for heartbeat replies until we reach self.time_to_dead.
118
95
119 Ignores interrupts, and returns the result of poll(), which
96 Ignores interrupts, and returns the result of poll(), which
120 will be an empty list if no messages arrived before the timeout,
97 will be an empty list if no messages arrived before the timeout,
121 or the event tuple if there is a message to receive.
98 or the event tuple if there is a message to receive.
122 """
99 """
123
100
124 until_dead = self.time_to_dead - (time.time() - start_time)
101 until_dead = self.time_to_dead - (time.time() - start_time)
125 # ensure poll at least once
102 # ensure poll at least once
126 until_dead = max(until_dead, 1e-3)
103 until_dead = max(until_dead, 1e-3)
127 events = []
104 events = []
128 while True:
105 while True:
129 try:
106 try:
130 events = self.poller.poll(1000 * until_dead)
107 events = self.poller.poll(1000 * until_dead)
131 except ZMQError as e:
108 except ZMQError as e:
132 if e.errno == errno.EINTR:
109 if e.errno == errno.EINTR:
133 # ignore interrupts during heartbeat
110 # ignore interrupts during heartbeat
134 # this may never actually happen
111 # this may never actually happen
135 until_dead = self.time_to_dead - (time.time() - start_time)
112 until_dead = self.time_to_dead - (time.time() - start_time)
136 until_dead = max(until_dead, 1e-3)
113 until_dead = max(until_dead, 1e-3)
137 pass
114 pass
138 else:
115 else:
139 raise
116 raise
140 except Exception:
117 except Exception:
141 if self._exiting:
118 if self._exiting:
142 break
119 break
143 else:
120 else:
144 raise
121 raise
145 else:
122 else:
146 break
123 break
147 return events
124 return events
148
125
149 def run(self):
126 def run(self):
150 """The thread's main activity. Call start() instead."""
127 """The thread's main activity. Call start() instead."""
151 self._create_socket()
128 self._create_socket()
152 self._running = True
129 self._running = True
153 self._beating = True
130 self._beating = True
154
131
155 while self._running:
132 while self._running:
156 if self._pause:
133 if self._pause:
157 # just sleep, and skip the rest of the loop
134 # just sleep, and skip the rest of the loop
158 time.sleep(self.time_to_dead)
135 time.sleep(self.time_to_dead)
159 continue
136 continue
160
137
161 since_last_heartbeat = 0.0
138 since_last_heartbeat = 0.0
162 # io.rprint('Ping from HB channel') # dbg
139 # io.rprint('Ping from HB channel') # dbg
163 # no need to catch EFSM here, because the previous event was
140 # no need to catch EFSM here, because the previous event was
164 # either a recv or connect, which cannot be followed by EFSM
141 # either a recv or connect, which cannot be followed by EFSM
165 self.socket.send(b'ping')
142 self.socket.send(b'ping')
166 request_time = time.time()
143 request_time = time.time()
167 ready = self._poll(request_time)
144 ready = self._poll(request_time)
168 if ready:
145 if ready:
169 self._beating = True
146 self._beating = True
170 # the poll above guarantees we have something to recv
147 # the poll above guarantees we have something to recv
171 self.socket.recv()
148 self.socket.recv()
172 # sleep the remainder of the cycle
149 # sleep the remainder of the cycle
173 remainder = self.time_to_dead - (time.time() - request_time)
150 remainder = self.time_to_dead - (time.time() - request_time)
174 if remainder > 0:
151 if remainder > 0:
175 time.sleep(remainder)
152 time.sleep(remainder)
176 continue
153 continue
177 else:
154 else:
178 # nothing was received within the time limit, signal heart failure
155 # nothing was received within the time limit, signal heart failure
179 self._beating = False
156 self._beating = False
180 since_last_heartbeat = time.time() - request_time
157 since_last_heartbeat = time.time() - request_time
181 self.call_handlers(since_last_heartbeat)
158 self.call_handlers(since_last_heartbeat)
182 # and close/reopen the socket, because the REQ/REP cycle has been broken
159 # and close/reopen the socket, because the REQ/REP cycle has been broken
183 self._create_socket()
160 self._create_socket()
184 continue
161 continue
185
162
186 def pause(self):
163 def pause(self):
187 """Pause the heartbeat."""
164 """Pause the heartbeat."""
188 self._pause = True
165 self._pause = True
189
166
190 def unpause(self):
167 def unpause(self):
191 """Unpause the heartbeat."""
168 """Unpause the heartbeat."""
192 self._pause = False
169 self._pause = False
193
170
194 def is_beating(self):
171 def is_beating(self):
195 """Is the heartbeat running and responsive (and not paused)."""
172 """Is the heartbeat running and responsive (and not paused)."""
196 if self.is_alive() and not self._pause and self._beating:
173 if self.is_alive() and not self._pause and self._beating:
197 return True
174 return True
198 else:
175 else:
199 return False
176 return False
200
177
201 def stop(self):
178 def stop(self):
202 """Stop the channel's event loop and join its thread."""
179 """Stop the channel's event loop and join its thread."""
203 self._running = False
180 self._running = False
204 self.join()
181 self.join()
205 self.close()
182 self.close()
206
183
207 def close(self):
184 def close(self):
208 if self.socket is not None:
185 if self.socket is not None:
209 try:
186 try:
210 self.socket.close(linger=0)
187 self.socket.close(linger=0)
211 except Exception:
188 except Exception:
212 pass
189 pass
213 self.socket = None
190 self.socket = None
214
191
215 def call_handlers(self, since_last_heartbeat):
192 def call_handlers(self, since_last_heartbeat):
216 """This method is called in the ioloop thread when a message arrives.
193 """This method is called in the ioloop thread when a message arrives.
217
194
218 Subclasses should override this method to handle incoming messages.
195 Subclasses should override this method to handle incoming messages.
219 It is important to remember that this method is called in the thread
196 It is important to remember that this method is called in the thread
220 so that some logic must be done to ensure that the application level
197 so that some logic must be done to ensure that the application level
221 handlers are called in the application thread.
198 handlers are called in the application thread.
222 """
199 """
223 pass
200 pass
224
201
225
202
226 HBChannelABC.register(HBChannel)
203 HBChannelABC.register(HBChannel)
@@ -1,389 +1,386 b''
1 """Base class to manage the interaction with a running kernel"""
1 """Base class to manage the interaction with a running kernel"""
2
2
3 # Copyright (c) IPython Development Team.
3 # Copyright (c) IPython Development Team.
4 # Distributed under the terms of the Modified BSD License.
4 # Distributed under the terms of the Modified BSD License.
5
5
6 from __future__ import absolute_import
6 from __future__ import absolute_import
7 from IPython.kernel.channels import major_protocol_version
7 from IPython.kernel.channels import major_protocol_version
8 from IPython.utils.py3compat import string_types, iteritems
8 from IPython.utils.py3compat import string_types, iteritems
9
9
10 import zmq
10 import zmq
11
11
12 from IPython.utils.traitlets import (
12 from IPython.utils.traitlets import (
13 Any, Instance, Type,
13 Any, Instance, Type,
14 )
14 )
15
15
16 from .channelsabc import (ChannelABC, HBChannelABC)
16 from .channelsabc import (ChannelABC, HBChannelABC)
17 from .channels import (
18 make_shell_socket, make_stdin_socket, make_iopub_socket
19 )
20 from .clientabc import KernelClientABC
17 from .clientabc import KernelClientABC
21 from .connect import ConnectionFileMixin
18 from .connect import ConnectionFileMixin
22
19
23
20
24 # some utilities to validate message structure, these might get moved elsewhere
21 # some utilities to validate message structure, these might get moved elsewhere
25 # if they prove to have more generic utility
22 # if they prove to have more generic utility
26
23
27 def validate_string_dict(dct):
24 def validate_string_dict(dct):
28 """Validate that the input is a dict with string keys and values.
25 """Validate that the input is a dict with string keys and values.
29
26
30 Raises ValueError if not."""
27 Raises ValueError if not."""
31 for k,v in iteritems(dct):
28 for k,v in iteritems(dct):
32 if not isinstance(k, string_types):
29 if not isinstance(k, string_types):
33 raise ValueError('key %r in dict must be a string' % k)
30 raise ValueError('key %r in dict must be a string' % k)
34 if not isinstance(v, string_types):
31 if not isinstance(v, string_types):
35 raise ValueError('value %r in dict must be a string' % v)
32 raise ValueError('value %r in dict must be a string' % v)
36
33
37
34
38 class KernelClient(ConnectionFileMixin):
35 class KernelClient(ConnectionFileMixin):
39 """Communicates with a single kernel on any host via zmq channels.
36 """Communicates with a single kernel on any host via zmq channels.
40
37
41 There are four channels associated with each kernel:
38 There are four channels associated with each kernel:
42
39
43 * shell: for request/reply calls to the kernel.
40 * shell: for request/reply calls to the kernel.
44 * iopub: for the kernel to publish results to frontends.
41 * iopub: for the kernel to publish results to frontends.
45 * hb: for monitoring the kernel's heartbeat.
42 * hb: for monitoring the kernel's heartbeat.
46 * stdin: for frontends to reply to raw_input calls in the kernel.
43 * stdin: for frontends to reply to raw_input calls in the kernel.
47
44
48 The methods of the channels are exposed as methods of the client itself
45 The methods of the channels are exposed as methods of the client itself
49 (KernelClient.execute, complete, history, etc.).
46 (KernelClient.execute, complete, history, etc.).
50 See the channels themselves for documentation of these methods.
47 See the channels themselves for documentation of these methods.
51
48
52 """
49 """
53
50
54 # The PyZMQ Context to use for communication with the kernel.
51 # The PyZMQ Context to use for communication with the kernel.
55 context = Instance(zmq.Context)
52 context = Instance(zmq.Context)
56 def _context_default(self):
53 def _context_default(self):
57 return zmq.Context.instance()
54 return zmq.Context.instance()
58
55
59 # The classes to use for the various channels
56 # The classes to use for the various channels
60 shell_channel_class = Type(ChannelABC)
57 shell_channel_class = Type(ChannelABC)
61 iopub_channel_class = Type(ChannelABC)
58 iopub_channel_class = Type(ChannelABC)
62 stdin_channel_class = Type(ChannelABC)
59 stdin_channel_class = Type(ChannelABC)
63 hb_channel_class = Type(HBChannelABC)
60 hb_channel_class = Type(HBChannelABC)
64
61
65 # Protected traits
62 # Protected traits
66 _shell_channel = Any
63 _shell_channel = Any
67 _iopub_channel = Any
64 _iopub_channel = Any
68 _stdin_channel = Any
65 _stdin_channel = Any
69 _hb_channel = Any
66 _hb_channel = Any
70
67
71 # flag for whether execute requests should be allowed to call raw_input:
68 # flag for whether execute requests should be allowed to call raw_input:
72 allow_stdin = True
69 allow_stdin = True
73
70
74 #--------------------------------------------------------------------------
71 #--------------------------------------------------------------------------
75 # Channel proxy methods
72 # Channel proxy methods
76 #--------------------------------------------------------------------------
73 #--------------------------------------------------------------------------
77
74
78 def _get_msg(channel, *args, **kwargs):
75 def _get_msg(channel, *args, **kwargs):
79 return channel.get_msg(*args, **kwargs)
76 return channel.get_msg(*args, **kwargs)
80
77
81 def get_shell_msg(self, *args, **kwargs):
78 def get_shell_msg(self, *args, **kwargs):
82 """Get a message from the shell channel"""
79 """Get a message from the shell channel"""
83 return self.shell_channel.get_msg(*args, **kwargs)
80 return self.shell_channel.get_msg(*args, **kwargs)
84
81
85 def get_iopub_msg(self, *args, **kwargs):
82 def get_iopub_msg(self, *args, **kwargs):
86 """Get a message from the iopub channel"""
83 """Get a message from the iopub channel"""
87 return self.iopub_channel.get_msg(*args, **kwargs)
84 return self.iopub_channel.get_msg(*args, **kwargs)
88
85
89 def get_stdin_msg(self, *args, **kwargs):
86 def get_stdin_msg(self, *args, **kwargs):
90 """Get a message from the stdin channel"""
87 """Get a message from the stdin channel"""
91 return self.stdin_channel.get_msg(*args, **kwargs)
88 return self.stdin_channel.get_msg(*args, **kwargs)
92
89
93 #--------------------------------------------------------------------------
90 #--------------------------------------------------------------------------
94 # Channel management methods
91 # Channel management methods
95 #--------------------------------------------------------------------------
92 #--------------------------------------------------------------------------
96
93
97 def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
94 def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
98 """Starts the channels for this kernel.
95 """Starts the channels for this kernel.
99
96
100 This will create the channels if they do not exist and then start
97 This will create the channels if they do not exist and then start
101 them (their activity runs in a thread). If port numbers of 0 are
98 them (their activity runs in a thread). If port numbers of 0 are
102 being used (random ports) then you must first call
99 being used (random ports) then you must first call
103 :meth:`start_kernel`. If the channels have been stopped and you
100 :meth:`start_kernel`. If the channels have been stopped and you
104 call this, :class:`RuntimeError` will be raised.
101 call this, :class:`RuntimeError` will be raised.
105 """
102 """
106 if shell:
103 if shell:
107 self.shell_channel.start()
104 self.shell_channel.start()
108 self.kernel_info()
105 self.kernel_info()
109 if iopub:
106 if iopub:
110 self.iopub_channel.start()
107 self.iopub_channel.start()
111 if stdin:
108 if stdin:
112 self.stdin_channel.start()
109 self.stdin_channel.start()
113 self.allow_stdin = True
110 self.allow_stdin = True
114 else:
111 else:
115 self.allow_stdin = False
112 self.allow_stdin = False
116 if hb:
113 if hb:
117 self.hb_channel.start()
114 self.hb_channel.start()
118
115
119 def stop_channels(self):
116 def stop_channels(self):
120 """Stops all the running channels for this kernel.
117 """Stops all the running channels for this kernel.
121
118
122 This stops their event loops and joins their threads.
119 This stops their event loops and joins their threads.
123 """
120 """
124 if self.shell_channel.is_alive():
121 if self.shell_channel.is_alive():
125 self.shell_channel.stop()
122 self.shell_channel.stop()
126 if self.iopub_channel.is_alive():
123 if self.iopub_channel.is_alive():
127 self.iopub_channel.stop()
124 self.iopub_channel.stop()
128 if self.stdin_channel.is_alive():
125 if self.stdin_channel.is_alive():
129 self.stdin_channel.stop()
126 self.stdin_channel.stop()
130 if self.hb_channel.is_alive():
127 if self.hb_channel.is_alive():
131 self.hb_channel.stop()
128 self.hb_channel.stop()
132
129
133 @property
130 @property
134 def channels_running(self):
131 def channels_running(self):
135 """Are any of the channels created and running?"""
132 """Are any of the channels created and running?"""
136 return (self.shell_channel.is_alive() or self.iopub_channel.is_alive() or
133 return (self.shell_channel.is_alive() or self.iopub_channel.is_alive() or
137 self.stdin_channel.is_alive() or self.hb_channel.is_alive())
134 self.stdin_channel.is_alive() or self.hb_channel.is_alive())
138
135
139 ioloop = None # Overridden in subclasses that use pyzmq event loop
136 ioloop = None # Overridden in subclasses that use pyzmq event loop
140
137
141 @property
138 @property
142 def shell_channel(self):
139 def shell_channel(self):
143 """Get the shell channel object for this kernel."""
140 """Get the shell channel object for this kernel."""
144 if self._shell_channel is None:
141 if self._shell_channel is None:
145 url = self._make_url('shell')
142 url = self._make_url('shell')
146 self.log.debug("connecting shell channel to %s", url)
143 self.log.debug("connecting shell channel to %s", url)
147 socket = make_shell_socket(self.context, self.session.bsession, url)
144 socket = self.connect_shell(identity=self.session.bsession)
148 self._shell_channel = self.shell_channel_class(
145 self._shell_channel = self.shell_channel_class(
149 socket, self.session, self.ioloop
146 socket, self.session, self.ioloop
150 )
147 )
151 return self._shell_channel
148 return self._shell_channel
152
149
153 @property
150 @property
154 def iopub_channel(self):
151 def iopub_channel(self):
155 """Get the iopub channel object for this kernel."""
152 """Get the iopub channel object for this kernel."""
156 if self._iopub_channel is None:
153 if self._iopub_channel is None:
157 url = self._make_url('iopub')
154 url = self._make_url('iopub')
158 self.log.debug("connecting iopub channel to %s", url)
155 self.log.debug("connecting iopub channel to %s", url)
159 socket = make_iopub_socket(self.context, self.session.bsession, url)
156 socket = self.connect_iopub()
160 self._iopub_channel = self.iopub_channel_class(
157 self._iopub_channel = self.iopub_channel_class(
161 socket, self.session, self.ioloop
158 socket, self.session, self.ioloop
162 )
159 )
163 return self._iopub_channel
160 return self._iopub_channel
164
161
165 @property
162 @property
166 def stdin_channel(self):
163 def stdin_channel(self):
167 """Get the stdin channel object for this kernel."""
164 """Get the stdin channel object for this kernel."""
168 if self._stdin_channel is None:
165 if self._stdin_channel is None:
169 url = self._make_url('stdin')
166 url = self._make_url('stdin')
170 self.log.debug("connecting stdin channel to %s", url)
167 self.log.debug("connecting stdin channel to %s", url)
171 socket = make_stdin_socket(self.context, self.session.bsession, url)
168 socket = self.connect_stdin(identity=self.session.bsession)
172 self._stdin_channel = self.stdin_channel_class(
169 self._stdin_channel = self.stdin_channel_class(
173 socket, self.session, self.ioloop
170 socket, self.session, self.ioloop
174 )
171 )
175 return self._stdin_channel
172 return self._stdin_channel
176
173
177 @property
174 @property
178 def hb_channel(self):
175 def hb_channel(self):
179 """Get the hb channel object for this kernel."""
176 """Get the hb channel object for this kernel."""
180 if self._hb_channel is None:
177 if self._hb_channel is None:
181 url = self._make_url('hb')
178 url = self._make_url('hb')
182 self.log.debug("connecting heartbeat channel to %s", url)
179 self.log.debug("connecting heartbeat channel to %s", url)
183 self._hb_channel = self.hb_channel_class(
180 self._hb_channel = self.hb_channel_class(
184 self.context, self.session, url
181 self.context, self.session, url
185 )
182 )
186 return self._hb_channel
183 return self._hb_channel
187
184
188 def is_alive(self):
185 def is_alive(self):
189 """Is the kernel process still running?"""
186 """Is the kernel process still running?"""
190 if self._hb_channel is not None:
187 if self._hb_channel is not None:
191 # We didn't start the kernel with this KernelManager so we
188 # We didn't start the kernel with this KernelManager so we
192 # use the heartbeat.
189 # use the heartbeat.
193 return self._hb_channel.is_beating()
190 return self._hb_channel.is_beating()
194 else:
191 else:
195 # no heartbeat and not local, we can't tell if it's running,
192 # no heartbeat and not local, we can't tell if it's running,
196 # so naively return True
193 # so naively return True
197 return True
194 return True
198
195
199
196
200 # Methods to send specific messages on channels
197 # Methods to send specific messages on channels
201 def execute(self, code, silent=False, store_history=True,
198 def execute(self, code, silent=False, store_history=True,
202 user_expressions=None, allow_stdin=None):
199 user_expressions=None, allow_stdin=None):
203 """Execute code in the kernel.
200 """Execute code in the kernel.
204
201
205 Parameters
202 Parameters
206 ----------
203 ----------
207 code : str
204 code : str
208 A string of Python code.
205 A string of Python code.
209
206
210 silent : bool, optional (default False)
207 silent : bool, optional (default False)
211 If set, the kernel will execute the code as quietly possible, and
208 If set, the kernel will execute the code as quietly possible, and
212 will force store_history to be False.
209 will force store_history to be False.
213
210
214 store_history : bool, optional (default True)
211 store_history : bool, optional (default True)
215 If set, the kernel will store command history. This is forced
212 If set, the kernel will store command history. This is forced
216 to be False if silent is True.
213 to be False if silent is True.
217
214
218 user_expressions : dict, optional
215 user_expressions : dict, optional
219 A dict mapping names to expressions to be evaluated in the user's
216 A dict mapping names to expressions to be evaluated in the user's
220 dict. The expression values are returned as strings formatted using
217 dict. The expression values are returned as strings formatted using
221 :func:`repr`.
218 :func:`repr`.
222
219
223 allow_stdin : bool, optional (default self.allow_stdin)
220 allow_stdin : bool, optional (default self.allow_stdin)
224 Flag for whether the kernel can send stdin requests to frontends.
221 Flag for whether the kernel can send stdin requests to frontends.
225
222
226 Some frontends (e.g. the Notebook) do not support stdin requests.
223 Some frontends (e.g. the Notebook) do not support stdin requests.
227 If raw_input is called from code executed from such a frontend, a
224 If raw_input is called from code executed from such a frontend, a
228 StdinNotImplementedError will be raised.
225 StdinNotImplementedError will be raised.
229
226
230 Returns
227 Returns
231 -------
228 -------
232 The msg_id of the message sent.
229 The msg_id of the message sent.
233 """
230 """
234 if user_expressions is None:
231 if user_expressions is None:
235 user_expressions = {}
232 user_expressions = {}
236 if allow_stdin is None:
233 if allow_stdin is None:
237 allow_stdin = self.allow_stdin
234 allow_stdin = self.allow_stdin
238
235
239
236
240 # Don't waste network traffic if inputs are invalid
237 # Don't waste network traffic if inputs are invalid
241 if not isinstance(code, string_types):
238 if not isinstance(code, string_types):
242 raise ValueError('code %r must be a string' % code)
239 raise ValueError('code %r must be a string' % code)
243 validate_string_dict(user_expressions)
240 validate_string_dict(user_expressions)
244
241
245 # Create class for content/msg creation. Related to, but possibly
242 # Create class for content/msg creation. Related to, but possibly
246 # not in Session.
243 # not in Session.
247 content = dict(code=code, silent=silent, store_history=store_history,
244 content = dict(code=code, silent=silent, store_history=store_history,
248 user_expressions=user_expressions,
245 user_expressions=user_expressions,
249 allow_stdin=allow_stdin,
246 allow_stdin=allow_stdin,
250 )
247 )
251 msg = self.session.msg('execute_request', content)
248 msg = self.session.msg('execute_request', content)
252 self.shell_channel.send(msg)
249 self.shell_channel.send(msg)
253 return msg['header']['msg_id']
250 return msg['header']['msg_id']
254
251
255 def complete(self, code, cursor_pos=None):
252 def complete(self, code, cursor_pos=None):
256 """Tab complete text in the kernel's namespace.
253 """Tab complete text in the kernel's namespace.
257
254
258 Parameters
255 Parameters
259 ----------
256 ----------
260 code : str
257 code : str
261 The context in which completion is requested.
258 The context in which completion is requested.
262 Can be anything between a variable name and an entire cell.
259 Can be anything between a variable name and an entire cell.
263 cursor_pos : int, optional
260 cursor_pos : int, optional
264 The position of the cursor in the block of code where the completion was requested.
261 The position of the cursor in the block of code where the completion was requested.
265 Default: ``len(code)``
262 Default: ``len(code)``
266
263
267 Returns
264 Returns
268 -------
265 -------
269 The msg_id of the message sent.
266 The msg_id of the message sent.
270 """
267 """
271 if cursor_pos is None:
268 if cursor_pos is None:
272 cursor_pos = len(code)
269 cursor_pos = len(code)
273 content = dict(code=code, cursor_pos=cursor_pos)
270 content = dict(code=code, cursor_pos=cursor_pos)
274 msg = self.session.msg('complete_request', content)
271 msg = self.session.msg('complete_request', content)
275 self.shell_channel.send(msg)
272 self.shell_channel.send(msg)
276 return msg['header']['msg_id']
273 return msg['header']['msg_id']
277
274
278 def inspect(self, code, cursor_pos=None, detail_level=0):
275 def inspect(self, code, cursor_pos=None, detail_level=0):
279 """Get metadata information about an object in the kernel's namespace.
276 """Get metadata information about an object in the kernel's namespace.
280
277
281 It is up to the kernel to determine the appropriate object to inspect.
278 It is up to the kernel to determine the appropriate object to inspect.
282
279
283 Parameters
280 Parameters
284 ----------
281 ----------
285 code : str
282 code : str
286 The context in which info is requested.
283 The context in which info is requested.
287 Can be anything between a variable name and an entire cell.
284 Can be anything between a variable name and an entire cell.
288 cursor_pos : int, optional
285 cursor_pos : int, optional
289 The position of the cursor in the block of code where the info was requested.
286 The position of the cursor in the block of code where the info was requested.
290 Default: ``len(code)``
287 Default: ``len(code)``
291 detail_level : int, optional
288 detail_level : int, optional
292 The level of detail for the introspection (0-2)
289 The level of detail for the introspection (0-2)
293
290
294 Returns
291 Returns
295 -------
292 -------
296 The msg_id of the message sent.
293 The msg_id of the message sent.
297 """
294 """
298 if cursor_pos is None:
295 if cursor_pos is None:
299 cursor_pos = len(code)
296 cursor_pos = len(code)
300 content = dict(code=code, cursor_pos=cursor_pos,
297 content = dict(code=code, cursor_pos=cursor_pos,
301 detail_level=detail_level,
298 detail_level=detail_level,
302 )
299 )
303 msg = self.session.msg('inspect_request', content)
300 msg = self.session.msg('inspect_request', content)
304 self.shell_channel.send(msg)
301 self.shell_channel.send(msg)
305 return msg['header']['msg_id']
302 return msg['header']['msg_id']
306
303
307 def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
304 def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
308 """Get entries from the kernel's history list.
305 """Get entries from the kernel's history list.
309
306
310 Parameters
307 Parameters
311 ----------
308 ----------
312 raw : bool
309 raw : bool
313 If True, return the raw input.
310 If True, return the raw input.
314 output : bool
311 output : bool
315 If True, then return the output as well.
312 If True, then return the output as well.
316 hist_access_type : str
313 hist_access_type : str
317 'range' (fill in session, start and stop params), 'tail' (fill in n)
314 'range' (fill in session, start and stop params), 'tail' (fill in n)
318 or 'search' (fill in pattern param).
315 or 'search' (fill in pattern param).
319
316
320 session : int
317 session : int
321 For a range request, the session from which to get lines. Session
318 For a range request, the session from which to get lines. Session
322 numbers are positive integers; negative ones count back from the
319 numbers are positive integers; negative ones count back from the
323 current session.
320 current session.
324 start : int
321 start : int
325 The first line number of a history range.
322 The first line number of a history range.
326 stop : int
323 stop : int
327 The final (excluded) line number of a history range.
324 The final (excluded) line number of a history range.
328
325
329 n : int
326 n : int
330 The number of lines of history to get for a tail request.
327 The number of lines of history to get for a tail request.
331
328
332 pattern : str
329 pattern : str
333 The glob-syntax pattern for a search request.
330 The glob-syntax pattern for a search request.
334
331
335 Returns
332 Returns
336 -------
333 -------
337 The msg_id of the message sent.
334 The msg_id of the message sent.
338 """
335 """
339 content = dict(raw=raw, output=output, hist_access_type=hist_access_type,
336 content = dict(raw=raw, output=output, hist_access_type=hist_access_type,
340 **kwargs)
337 **kwargs)
341 msg = self.session.msg('history_request', content)
338 msg = self.session.msg('history_request', content)
342 self.shell_channel.send(msg)
339 self.shell_channel.send(msg)
343 return msg['header']['msg_id']
340 return msg['header']['msg_id']
344
341
345 def kernel_info(self):
342 def kernel_info(self):
346 """Request kernel info."""
343 """Request kernel info."""
347 msg = self.session.msg('kernel_info_request')
344 msg = self.session.msg('kernel_info_request')
348 self.shell_channel.send(msg)
345 self.shell_channel.send(msg)
349 return msg['header']['msg_id']
346 return msg['header']['msg_id']
350
347
351 def _handle_kernel_info_reply(self, msg):
348 def _handle_kernel_info_reply(self, msg):
352 """handle kernel info reply
349 """handle kernel info reply
353
350
354 sets protocol adaptation version
351 sets protocol adaptation version
355 """
352 """
356 adapt_version = int(msg['content']['protocol_version'].split('.')[0])
353 adapt_version = int(msg['content']['protocol_version'].split('.')[0])
357 if adapt_version != major_protocol_version:
354 if adapt_version != major_protocol_version:
358 self.session.adapt_version = adapt_version
355 self.session.adapt_version = adapt_version
359
356
360 def shutdown(self, restart=False):
357 def shutdown(self, restart=False):
361 """Request an immediate kernel shutdown.
358 """Request an immediate kernel shutdown.
362
359
363 Upon receipt of the (empty) reply, client code can safely assume that
360 Upon receipt of the (empty) reply, client code can safely assume that
364 the kernel has shut down and it's safe to forcefully terminate it if
361 the kernel has shut down and it's safe to forcefully terminate it if
365 it's still alive.
362 it's still alive.
366
363
367 The kernel will send the reply via a function registered with Python's
364 The kernel will send the reply via a function registered with Python's
368 atexit module, ensuring it's truly done as the kernel is done with all
365 atexit module, ensuring it's truly done as the kernel is done with all
369 normal operation.
366 normal operation.
370 """
367 """
371 # Send quit message to kernel. Once we implement kernel-side setattr,
368 # Send quit message to kernel. Once we implement kernel-side setattr,
372 # this should probably be done that way, but for now this will do.
369 # this should probably be done that way, but for now this will do.
373 msg = self.session.msg('shutdown_request', {'restart':restart})
370 msg = self.session.msg('shutdown_request', {'restart':restart})
374 self.shell_channel.send(msg)
371 self.shell_channel.send(msg)
375 return msg['header']['msg_id']
372 return msg['header']['msg_id']
376
373
377 def is_complete(self, code):
374 def is_complete(self, code):
378 msg = self.session.msg('is_complete_request', {'code': code})
375 msg = self.session.msg('is_complete_request', {'code': code})
379 self.shell_channel.send(msg)
376 self.shell_channel.send(msg)
380 return msg['header']['msg_id']
377 return msg['header']['msg_id']
381
378
382 def input(self, string):
379 def input(self, string):
383 """Send a string of raw input to the kernel."""
380 """Send a string of raw input to the kernel."""
384 content = dict(value=string)
381 content = dict(value=string)
385 msg = self.session.msg('input_reply', content)
382 msg = self.session.msg('input_reply', content)
386 self.stdin_channel.send(msg)
383 self.stdin_channel.send(msg)
387
384
388
385
389 KernelClientABC.register(KernelClient)
386 KernelClientABC.register(KernelClient)
@@ -1,250 +1,249 b''
1 """ Defines a KernelClient that provides signals and slots.
1 """ Defines a KernelClient that provides signals and slots.
2 """
2 """
3 import atexit
3 import atexit
4 import errno
4 import errno
5 from threading import Thread
5 from threading import Thread
6 import time
6 import time
7
7
8 import zmq
8 import zmq
9 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
9 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
10 # during garbage collection of threads at exit:
10 # during garbage collection of threads at exit:
11 from zmq import ZMQError
11 from zmq import ZMQError
12 from zmq.eventloop import ioloop, zmqstream
12 from zmq.eventloop import ioloop, zmqstream
13
13
14 from IPython.external.qt import QtCore
14 from IPython.external.qt import QtCore
15
15
16 # Local imports
16 # Local imports
17 from IPython.utils.traitlets import Type, Instance
17 from IPython.utils.traitlets import Type, Instance
18 from IPython.kernel.channels import HBChannel,\
18 from IPython.kernel.channels import HBChannel
19 make_shell_socket, make_iopub_socket, make_stdin_socket
20 from IPython.kernel import KernelClient
19 from IPython.kernel import KernelClient
21
20
22 from .kernel_mixins import QtKernelClientMixin
21 from .kernel_mixins import QtKernelClientMixin
23 from .util import SuperQObject
22 from .util import SuperQObject
24
23
25 class QtHBChannel(SuperQObject, HBChannel):
24 class QtHBChannel(SuperQObject, HBChannel):
26 # A longer timeout than the base class
25 # A longer timeout than the base class
27 time_to_dead = 3.0
26 time_to_dead = 3.0
28
27
29 # Emitted when the kernel has died.
28 # Emitted when the kernel has died.
30 kernel_died = QtCore.Signal(object)
29 kernel_died = QtCore.Signal(object)
31
30
32 def call_handlers(self, since_last_heartbeat):
31 def call_handlers(self, since_last_heartbeat):
33 """ Reimplemented to emit signals instead of making callbacks.
32 """ Reimplemented to emit signals instead of making callbacks.
34 """
33 """
35 # Emit the generic signal.
34 # Emit the generic signal.
36 self.kernel_died.emit(since_last_heartbeat)
35 self.kernel_died.emit(since_last_heartbeat)
37
36
38 from IPython.core.release import kernel_protocol_version_info
37 from IPython.core.release import kernel_protocol_version_info
39
38
40 major_protocol_version = kernel_protocol_version_info[0]
39 major_protocol_version = kernel_protocol_version_info[0]
41
40
42 class InvalidPortNumber(Exception):
41 class InvalidPortNumber(Exception):
43 pass
42 pass
44
43
45
44
46 class QtZMQSocketChannel(SuperQObject):
45 class QtZMQSocketChannel(SuperQObject):
47 """A ZMQ socket emitting a Qt signal when a message is received."""
46 """A ZMQ socket emitting a Qt signal when a message is received."""
48 session = None
47 session = None
49 socket = None
48 socket = None
50 ioloop = None
49 ioloop = None
51 stream = None
50 stream = None
52
51
53 message_received = QtCore.Signal(object)
52 message_received = QtCore.Signal(object)
54
53
55 def process_events(self):
54 def process_events(self):
56 """ Process any pending GUI events.
55 """ Process any pending GUI events.
57 """
56 """
58 QtCore.QCoreApplication.instance().processEvents()
57 QtCore.QCoreApplication.instance().processEvents()
59
58
60 def __init__(self, socket, session, loop):
59 def __init__(self, socket, session, loop):
61 """Create a channel.
60 """Create a channel.
62
61
63 Parameters
62 Parameters
64 ----------
63 ----------
65 socket : :class:`zmq.Socket`
64 socket : :class:`zmq.Socket`
66 The ZMQ socket to use.
65 The ZMQ socket to use.
67 session : :class:`session.Session`
66 session : :class:`session.Session`
68 The session to use.
67 The session to use.
69 loop
68 loop
70 A pyzmq ioloop to connect the socket to using a ZMQStream
69 A pyzmq ioloop to connect the socket to using a ZMQStream
71 """
70 """
72 super(QtZMQSocketChannel, self).__init__()
71 super(QtZMQSocketChannel, self).__init__()
73
72
74 self.socket = socket
73 self.socket = socket
75 self.session = session
74 self.session = session
76 self.ioloop = loop
75 self.ioloop = loop
77
76
78 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
77 self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
79 self.stream.on_recv(self._handle_recv)
78 self.stream.on_recv(self._handle_recv)
80
79
81 _is_alive = False
80 _is_alive = False
82 def is_alive(self):
81 def is_alive(self):
83 return self._is_alive
82 return self._is_alive
84
83
85 def start(self):
84 def start(self):
86 self._is_alive = True
85 self._is_alive = True
87
86
88 def stop(self):
87 def stop(self):
89 self._is_alive = False
88 self._is_alive = False
90
89
91 def close(self):
90 def close(self):
92 if self.socket is not None:
91 if self.socket is not None:
93 try:
92 try:
94 self.socket.close(linger=0)
93 self.socket.close(linger=0)
95 except Exception:
94 except Exception:
96 pass
95 pass
97 self.socket = None
96 self.socket = None
98
97
99 def send(self, msg):
98 def send(self, msg):
100 """Queue a message to be sent from the IOLoop's thread.
99 """Queue a message to be sent from the IOLoop's thread.
101
100
102 Parameters
101 Parameters
103 ----------
102 ----------
104 msg : message to send
103 msg : message to send
105
104
106 This is threadsafe, as it uses IOLoop.add_callback to give the loop's
105 This is threadsafe, as it uses IOLoop.add_callback to give the loop's
107 thread control of the action.
106 thread control of the action.
108 """
107 """
109 def thread_send():
108 def thread_send():
110 self.session.send(self.stream, msg)
109 self.session.send(self.stream, msg)
111 self.ioloop.add_callback(thread_send)
110 self.ioloop.add_callback(thread_send)
112
111
113 def _handle_recv(self, msg):
112 def _handle_recv(self, msg):
114 """Callback for stream.on_recv.
113 """Callback for stream.on_recv.
115
114
116 Unpacks message, and calls handlers with it.
115 Unpacks message, and calls handlers with it.
117 """
116 """
118 ident,smsg = self.session.feed_identities(msg)
117 ident,smsg = self.session.feed_identities(msg)
119 msg = self.session.deserialize(smsg)
118 msg = self.session.deserialize(smsg)
120 self.call_handlers(msg)
119 self.call_handlers(msg)
121
120
122 def call_handlers(self, msg):
121 def call_handlers(self, msg):
123 """This method is called in the ioloop thread when a message arrives.
122 """This method is called in the ioloop thread when a message arrives.
124
123
125 Subclasses should override this method to handle incoming messages.
124 Subclasses should override this method to handle incoming messages.
126 It is important to remember that this method is called in the thread
125 It is important to remember that this method is called in the thread
127 so that some logic must be done to ensure that the application level
126 so that some logic must be done to ensure that the application level
128 handlers are called in the application thread.
127 handlers are called in the application thread.
129 """
128 """
130 # Emit the generic signal.
129 # Emit the generic signal.
131 self.message_received.emit(msg)
130 self.message_received.emit(msg)
132
131
133 def flush(self, timeout=1.0):
132 def flush(self, timeout=1.0):
134 """Immediately processes all pending messages on this channel.
133 """Immediately processes all pending messages on this channel.
135
134
136 This is only used for the IOPub channel.
135 This is only used for the IOPub channel.
137
136
138 Callers should use this method to ensure that :meth:`call_handlers`
137 Callers should use this method to ensure that :meth:`call_handlers`
139 has been called for all messages that have been received on the
138 has been called for all messages that have been received on the
140 0MQ SUB socket of this channel.
139 0MQ SUB socket of this channel.
141
140
142 This method is thread safe.
141 This method is thread safe.
143
142
144 Parameters
143 Parameters
145 ----------
144 ----------
146 timeout : float, optional
145 timeout : float, optional
147 The maximum amount of time to spend flushing, in seconds. The
146 The maximum amount of time to spend flushing, in seconds. The
148 default is one second.
147 default is one second.
149 """
148 """
150 # We do the IOLoop callback process twice to ensure that the IOLoop
149 # We do the IOLoop callback process twice to ensure that the IOLoop
151 # gets to perform at least one full poll.
150 # gets to perform at least one full poll.
152 stop_time = time.time() + timeout
151 stop_time = time.time() + timeout
153 for i in range(2):
152 for i in range(2):
154 self._flushed = False
153 self._flushed = False
155 self.ioloop.add_callback(self._flush)
154 self.ioloop.add_callback(self._flush)
156 while not self._flushed and time.time() < stop_time:
155 while not self._flushed and time.time() < stop_time:
157 time.sleep(0.01)
156 time.sleep(0.01)
158
157
159 def _flush(self):
158 def _flush(self):
160 """Callback for :method:`self.flush`."""
159 """Callback for :method:`self.flush`."""
161 self.stream.flush()
160 self.stream.flush()
162 self._flushed = True
161 self._flushed = True
163
162
164
163
165 class IOLoopThread(Thread):
164 class IOLoopThread(Thread):
166 """Run a pyzmq ioloop in a thread to send and receive messages
165 """Run a pyzmq ioloop in a thread to send and receive messages
167 """
166 """
168 def __init__(self, loop):
167 def __init__(self, loop):
169 super(IOLoopThread, self).__init__()
168 super(IOLoopThread, self).__init__()
170 self.daemon = True
169 self.daemon = True
171 atexit.register(self._notice_exit)
170 atexit.register(self._notice_exit)
172 self.ioloop = loop or ioloop.IOLoop()
171 self.ioloop = loop or ioloop.IOLoop()
173
172
174 def _notice_exit(self):
173 def _notice_exit(self):
175 self._exiting = True
174 self._exiting = True
176
175
177 def run(self):
176 def run(self):
178 """Run my loop, ignoring EINTR events in the poller"""
177 """Run my loop, ignoring EINTR events in the poller"""
179 while True:
178 while True:
180 try:
179 try:
181 self.ioloop.start()
180 self.ioloop.start()
182 except ZMQError as e:
181 except ZMQError as e:
183 if e.errno == errno.EINTR:
182 if e.errno == errno.EINTR:
184 continue
183 continue
185 else:
184 else:
186 raise
185 raise
187 except Exception:
186 except Exception:
188 if self._exiting:
187 if self._exiting:
189 break
188 break
190 else:
189 else:
191 raise
190 raise
192 else:
191 else:
193 break
192 break
194
193
195 def stop(self):
194 def stop(self):
196 """Stop the channel's event loop and join its thread.
195 """Stop the channel's event loop and join its thread.
197
196
198 This calls :meth:`~threading.Thread.join` and returns when the thread
197 This calls :meth:`~threading.Thread.join` and returns when the thread
199 terminates. :class:`RuntimeError` will be raised if
198 terminates. :class:`RuntimeError` will be raised if
200 :meth:`~threading.Thread.start` is called again.
199 :meth:`~threading.Thread.start` is called again.
201 """
200 """
202 if self.ioloop is not None:
201 if self.ioloop is not None:
203 self.ioloop.stop()
202 self.ioloop.stop()
204 self.join()
203 self.join()
205 self.close()
204 self.close()
206
205
207 def close(self):
206 def close(self):
208 if self.ioloop is not None:
207 if self.ioloop is not None:
209 try:
208 try:
210 self.ioloop.close(all_fds=True)
209 self.ioloop.close(all_fds=True)
211 except Exception:
210 except Exception:
212 pass
211 pass
213
212
214
213
215 class QtKernelClient(QtKernelClientMixin, KernelClient):
214 class QtKernelClient(QtKernelClientMixin, KernelClient):
216 """ A KernelClient that provides signals and slots.
215 """ A KernelClient that provides signals and slots.
217 """
216 """
218
217
219 _ioloop = None
218 _ioloop = None
220 @property
219 @property
221 def ioloop(self):
220 def ioloop(self):
222 if self._ioloop is None:
221 if self._ioloop is None:
223 self._ioloop = ioloop.IOLoop()
222 self._ioloop = ioloop.IOLoop()
224 return self._ioloop
223 return self._ioloop
225
224
226 ioloop_thread = Instance(IOLoopThread)
225 ioloop_thread = Instance(IOLoopThread)
227
226
228 def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
227 def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
229 if shell:
228 if shell:
230 self.shell_channel.message_received.connect(self._check_kernel_info_reply)
229 self.shell_channel.message_received.connect(self._check_kernel_info_reply)
231
230
232 self.ioloop_thread = IOLoopThread(self.ioloop)
231 self.ioloop_thread = IOLoopThread(self.ioloop)
233 self.ioloop_thread.start()
232 self.ioloop_thread.start()
234
233
235 super(QtKernelClient, self).start_channels(shell, iopub, stdin, hb)
234 super(QtKernelClient, self).start_channels(shell, iopub, stdin, hb)
236
235
237 def _check_kernel_info_reply(self, msg):
236 def _check_kernel_info_reply(self, msg):
238 if msg['msg_type'] == 'kernel_info_reply':
237 if msg['msg_type'] == 'kernel_info_reply':
239 self._handle_kernel_info_reply(msg)
238 self._handle_kernel_info_reply(msg)
240 self.shell_channel.message_received.disconnect(self._check_kernel_info_reply)
239 self.shell_channel.message_received.disconnect(self._check_kernel_info_reply)
241
240
242 def stop_channels(self):
241 def stop_channels(self):
243 super(QtKernelClient, self).stop_channels()
242 super(QtKernelClient, self).stop_channels()
244 if self.ioloop_thread.is_alive():
243 if self.ioloop_thread.is_alive():
245 self.ioloop_thread.stop()
244 self.ioloop_thread.stop()
246
245
247 iopub_channel_class = Type(QtZMQSocketChannel)
246 iopub_channel_class = Type(QtZMQSocketChannel)
248 shell_channel_class = Type(QtZMQSocketChannel)
247 shell_channel_class = Type(QtZMQSocketChannel)
249 stdin_channel_class = Type(QtZMQSocketChannel)
248 stdin_channel_class = Type(QtZMQSocketChannel)
250 hb_channel_class = Type(QtHBChannel)
249 hb_channel_class = Type(QtHBChannel)
General Comments 0
You need to be logged in to leave comments. Login now