@@ -138,7 +138,7 @@ def nudge(self):
138
138
"""Nudge the zmq connections with kernel_info_requests
139
139
140
140
Returns a Future that will resolve when we have received
141
- a shell reply and at least one iopub message,
141
+ a shell or control reply and at least one iopub message,
142
142
ensuring that zmq subscriptions are established,
143
143
sockets are fully connected, and kernel is responsive.
144
144
@@ -157,10 +157,12 @@ def nudge(self):
157
157
f = Future ()
158
158
f .set_result (None )
159
159
return f
160
-
161
160
# Use a transient shell channel to prevent leaking
162
161
# shell responses to the front-end.
163
162
shell_channel = kernel .connect_shell ()
163
+ # Use a transient control channel to prevent leaking
164
+ # control responses to the front-end.
165
+ control_channel = kernel .connect_control ()
164
166
# The IOPub used by the client, whose subscriptions we are verifying.
165
167
iopub_channel = self .channels ["iopub" ]
166
168
@@ -183,6 +185,8 @@ def cleanup(f=None):
183
185
iopub_channel .stop_on_recv ()
184
186
if not shell_channel .closed ():
185
187
shell_channel .close ()
188
+ if not control_channel .closed ():
189
+ control_channel .close ()
186
190
187
191
# trigger cleanup when both message futures are resolved
188
192
both_done .add_done_callback (cleanup )
@@ -193,6 +197,12 @@ def on_shell_reply(msg):
193
197
self .log .debug ("Nudge: resolving shell future: %s" , self .kernel_id )
194
198
info_future .set_result (None )
195
199
200
+ def on_control_reply (msg ):
201
+ self .log .debug ("Nudge: control info reply received: %s" , self .kernel_id )
202
+ if not info_future .done ():
203
+ self .log .debug ("Nudge: resolving control future: %s" , self .kernel_id )
204
+ info_future .set_result (None )
205
+
196
206
def on_iopub (msg ):
197
207
self .log .debug ("Nudge: IOPub received: %s" , self .kernel_id )
198
208
if not iopub_future .done ():
@@ -202,6 +212,7 @@ def on_iopub(msg):
202
212
203
213
iopub_channel .on_recv (on_iopub )
204
214
shell_channel .on_recv (on_shell_reply )
215
+ control_channel .on_recv (on_control_reply )
205
216
loop = IOLoop .current ()
206
217
207
218
# Nudge the kernel with kernel info requests until we get an IOPub message
@@ -227,6 +238,12 @@ def nudge(count):
227
238
228
239
# check for closed zmq socket
229
240
if shell_channel .closed ():
241
+ self .log .debug ("Nudge: cancelling on closed zmq socket: %s" , self .kernel_id )
242
+ finish ()
243
+ return
244
+
245
+ # check for closed zmq socket
246
+ if control_channel .closed ():
230
247
self .log .debug (
231
248
"Nudge: cancelling on closed zmq socket: %s" , self .kernel_id
232
249
)
@@ -237,6 +254,7 @@ def nudge(count):
237
254
log = self .log .warning if count % 10 == 0 else self .log .debug
238
255
log ("Nudge: attempt %s on kernel %s" % (count , self .kernel_id ))
239
256
self .session .send (shell_channel , "kernel_info_request" )
257
+ self .session .send (control_channel , "kernel_info_request" )
240
258
nonlocal nudge_handle
241
259
nudge_handle = loop .call_later (0.5 , nudge , count )
242
260
0 commit comments