Skip to content

Commit c30324d

Browse files
committed
Reply with busy -> error -> idle if subshell unknown
1 parent ec804fb commit c30324d

2 files changed

Lines changed: 88 additions & 2 deletions

File tree

‎ipykernel/kernelbase.py‎

Lines changed: 46 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -594,14 +594,18 @@ async def shell_channel_thread_main(self, msg):
594594

595595
# deserialize only the header to get subshell_id
596596
# Keep original message to send to subshell_id unmodified.
597-
_, msg2 = self.session.feed_identities(msg, copy=False)
597+
idents, msg2 = self.session.feed_identities(msg, copy=False)
598598
try:
599599
msg3 = self.session.deserialize(msg2, content=False, copy=False)
600600
subshell_id = msg3["header"].get("subshell_id")
601601

602602
# Find inproc pair socket to use to send message to correct subshell.
603603
subshell_manager = self.shell_channel_thread.manager
604-
socket = subshell_manager.get_shell_channel_to_subshell_socket(subshell_id)
604+
try:
605+
socket = subshell_manager.get_shell_channel_to_subshell_socket(subshell_id)
606+
except KeyError:
607+
self._send_unknown_subshell_reply(idents, msg3, subshell_id)
608+
return
605609
assert socket is not None
606610
socket.send_multipart(msg, copy=False)
607611
except Exception:
@@ -1378,6 +1382,46 @@ def _send_abort_reply(self, stream, msg, idents):
13781382
ident=idents,
13791383
)
13801384

1385+
def _send_unknown_subshell_reply(self, idents, msg, subshell_id) -> None:
1386+
"""Send an error reply to a request addressed to a subshell that is not there.
1387+
1388+
Runs in the shell channel thread, so it writes to the shell socket
1389+
directly instead of going through a subshell.
1390+
1391+
The busy and idle status messages matter as much as the reply here.
1392+
A client tracks the completion of a request by the idle status that
1393+
carries it as parent, and for message types that have no reply, such as
1394+
the comm messages, that status is all it has to go on.
1395+
"""
1396+
if not self.session:
1397+
return
1398+
msg_type = msg["header"]["msg_type"]
1399+
self.log.warning(
1400+
"Cannot handle %s %s: unknown subshell_id %r",
1401+
msg_type,
1402+
msg["header"]["msg_id"],
1403+
subshell_id,
1404+
)
1405+
self._publish_status("busy", "shell", parent=msg)
1406+
content = {
1407+
"status": "error",
1408+
"ename": "KeyError",
1409+
"evalue": f"Unknown subshell_id {subshell_id!r}",
1410+
"traceback": [],
1411+
}
1412+
md = self.init_metadata(msg)
1413+
md = self.finish_metadata(msg, md, content)
1414+
md.update({"status": "error"})
1415+
self.session.send(
1416+
self.shell_stream,
1417+
msg_type.rsplit("_", 1)[0] + "_reply",
1418+
metadata=md,
1419+
content=content,
1420+
parent=msg,
1421+
ident=idents,
1422+
)
1423+
self._publish_status("idle", "shell", parent=msg)
1424+
13811425
def _no_raw_input(self):
13821426
"""Raise StdinNotImplementedError if active frontend doesn't support
13831427
stdin."""

‎tests/test_subshells.py‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -391,3 +391,45 @@ def test_silent_flag_in_subshells():
391391
# Ensure subshell is always deleted
392392
if subshell_id:
393393
delete_subshell_helper(kc, subshell_id)
394+
395+
396+
def test_unknown_subshell_id():
397+
# A request for a subshell that does not exist is answered with an error rather
398+
# than dropped, so that a client waiting for it does not wait forever.
399+
with new_kernel() as kc:
400+
subshell_id = create_subshell_helper(kc)["subshell_id"]
401+
delete_subshell_helper(kc, subshell_id)
402+
flush_channels(kc)
403+
404+
msg = execute_request(kc, "a = 1", subshell_id)
405+
msg_id = msg["header"]["msg_id"]
406+
407+
reply = get_reply(kc, msg_id, TIMEOUT)
408+
assert reply["content"]["status"] == "error"
409+
assert subshell_id in reply["content"]["evalue"]
410+
411+
states = []
412+
while True:
413+
iopub_msg = kc.get_iopub_msg(timeout=TIMEOUT)
414+
if iopub_msg["parent_header"].get("msg_id") != msg_id:
415+
continue
416+
assert iopub_msg["msg_type"] == "status"
417+
states.append(iopub_msg["content"]["execution_state"])
418+
if states[-1] == "idle":
419+
break
420+
assert states == ["busy", "idle"]
421+
422+
423+
def test_comm_close_on_deleted_subshell():
424+
# A comm message has no reply of its own, so the idle status is the only thing
425+
# that tells the client the kernel is done with it.
426+
with new_kernel() as kc:
427+
subshell_id = create_subshell_helper(kc)["subshell_id"]
428+
delete_subshell_helper(kc, subshell_id)
429+
flush_channels(kc)
430+
431+
msg = kc.session.msg("comm_close", {"comm_id": "comm-1", "data": {}})
432+
msg["header"]["subshell_id"] = subshell_id
433+
kc.shell_channel.send(msg)
434+
435+
wait_for_idle(kc, msg["header"]["msg_id"])

0 commit comments

Comments
 (0)