Correct use of asyncio.Lock to process a single control message at a time (#1416)

Author: ianthomas23

ipykernel/kernelbase.py

@@ -246,6 +246,9 @@ def _parent_header(self):
     # execution count we store in the shell.
     execution_count = 0
 
+    # Asyncio lock to ensure only one control queue message is processed at a time.
+    _control_lock = Instance(asyncio.Lock)
+
     msg_types = [
         "execute_request",
         "complete_request",
@@ -295,7 +298,7 @@ def __init__(self, **kwargs):
 
     async def dispatch_control(self, msg):
         # Ensure only one control message is processed at a time
-        async with asyncio.Lock():
+        async with self._control_lock:
             await self.process_control(msg)
 
     async def process_control(self, msg):
@@ -540,6 +543,10 @@ def schedule_dispatch(self, dispatch, *args):
         # ensure the eventloop wakes up
         self.io_loop.add_callback(lambda: None)
 
+    async def _create_control_lock(self):
+        # This can be removed when minimum python increases to 3.10
+        self._control_lock = asyncio.Lock()
+
     def start(self):
         """register dispatchers for streams"""
         self.io_loop = ioloop.IOLoop.current()
@@ -549,6 +556,14 @@ def start(self):
         if self.control_stream:
             self.control_stream.on_recv(self.dispatch_control, copy=False)
 
+        if self.control_thread and sys.version_info < (3, 10):
+            # Before Python 3.10 we need to ensure the _control_lock is created in the
+            # thread that uses it. When our minimum python is 3.10 we can remove this
+            # and always use the else below, or just assign it where it is declared.
+            self.control_thread.io_loop.add_callback(self._create_control_lock)
+        else:
+            self._control_lock = asyncio.Lock()
+
         if self.shell_stream:
             self.shell_stream.on_recv(
                 partial(

tests/test_debugger.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from .utils import TIMEOUT, get_reply, new_kernel
+from .utils import TIMEOUT, get_replies, get_reply, new_kernel
 
 seq = 0
 
@@ -15,11 +15,8 @@
     debugpy = None
 
 
-def wait_for_debug_request(kernel, command, arguments=None, full_reply=False):
-    """Carry out a debug request and return the reply content.
-
-    It does not check if the request was successful.
-    """
+def prepare_debug_request(kernel, command, arguments=None):
+    """Prepare a debug request but do not send it."""
     global seq
     seq += 1
 
@@ -32,6 +29,15 @@ def wait_for_debug_request(kernel, command, arguments=None, full_reply=False):
             "arguments": arguments or {},
         },
     )
+    return msg
+
+
+def wait_for_debug_request(kernel, command, arguments=None, full_reply=False):
+    """Carry out a debug request and return the reply content.
+
+    It does not check if the request was successful.
+    """
+    msg = prepare_debug_request(kernel, command, arguments)
     kernel.control_channel.send(msg)
     reply = get_reply(kernel, msg["header"]["msg_id"], channel="control")
     return reply if full_reply else reply["content"]
@@ -448,3 +454,96 @@ def my_test():
 
     # Compare local and global variable
     assert global_var["value"] == local_var["value"] and global_var["type"] == local_var["type"]  # noqa: PT018
+
+
+def test_debug_requests_sequential(kernel_with_debug):
+    # Issue https://github.com/ipython/ipykernel/issues/1412
+    # Control channel requests should be executed sequentially not concurrently.
+    code = """def f(a, b):
+    c = a + b
+    return c
+
+f(2, 3)"""
+
+    r = wait_for_debug_request(kernel_with_debug, "dumpCell", {"code": code})
+    if debugpy:
+        source = r["body"]["sourcePath"]
+    else:
+        assert r == {}
+        source = "some path"
+
+    wait_for_debug_request(
+        kernel_with_debug,
+        "setBreakpoints",
+        {
+            "breakpoints": [{"line": 2}],
+            "source": {"path": source},
+            "sourceModified": False,
+        },
+    )
+
+    wait_for_debug_request(kernel_with_debug, "debugInfo")
+    wait_for_debug_request(kernel_with_debug, "configurationDone")
+    kernel_with_debug.execute(code)
+
+    if not debugpy:
+        # Cannot stop on breakpoint if debugpy not installed
+        return
+
+    # Wait for stop on breakpoint
+    msg: dict = {"msg_type": "", "content": {}}
+    while msg.get("msg_type") != "debug_event" or msg["content"].get("event") != "stopped":
+        msg = kernel_with_debug.get_iopub_msg(timeout=TIMEOUT)
+
+    stacks = wait_for_debug_request(kernel_with_debug, "stackTrace", {"threadId": 1})["body"][
+        "stackFrames"
+    ]
+
+    scopes = wait_for_debug_request(kernel_with_debug, "scopes", {"frameId": stacks[0]["id"]})[
+        "body"
+    ]["scopes"]
+
+    # Get variablesReference for both Locals and Globals.
+    locals_ref = next(filter(lambda s: s["name"] == "Locals", scopes))["variablesReference"]
+    globals_ref = next(filter(lambda s: s["name"] == "Globals", scopes))["variablesReference"]
+
+    msgs = []
+    for ref in [locals_ref, globals_ref]:
+        msgs.append(
+            prepare_debug_request(kernel_with_debug, "variables", {"variablesReference": ref})
+        )
+
+    # Send messages in quick succession.
+    for msg in msgs:
+        kernel_with_debug.control_channel.send(msg)
+
+    replies = get_replies(kernel_with_debug, [msg["msg_id"] for msg in msgs], channel="control")
+
+    # Check debug variable returns are correct.
+    locals = replies[0]["content"]
+    assert locals["success"]
+    variables = locals["body"]["variables"]
+    var = next(filter(lambda v: v["name"] == "a", variables))
+    assert var["type"] == "int"
+    assert var["value"] == "2"
+    var = next(filter(lambda v: v["name"] == "b", variables))
+    assert var["type"] == "int"
+    assert var["value"] == "3"
+
+    globals = replies[1]["content"]
+    assert globals["success"]
+    variables = globals["body"]["variables"]
+
+    names = [v["name"] for v in variables]
+    assert "function variables" in names
+    assert "special variables" in names
+
+    # Check status iopub messages alternate between busy and idle.
+    execution_states = []
+    while len(execution_states) < 8:
+        msg = kernel_with_debug.get_iopub_msg(timeout=TIMEOUT)
+        if msg["msg_type"] == "status":
+            execution_states.append(msg["content"]["execution_state"])
+    assert execution_states.count("busy") == 4
+    assert execution_states.count("idle") == 4
+    assert execution_states == ["busy", "idle"] * 4

tests/utils.py

@@ -62,6 +62,24 @@ def get_reply(kc, msg_id, timeout=TIMEOUT, channel="shell"):
     return reply
 
 
+def get_replies(kc, msg_ids: list[str], timeout=TIMEOUT, channel="shell"):
+    # Get replies which may arrive in any order as they may be running on different subshells.
+    # Replies are returned in the same order as the msg_ids, not in the order of arrival.
+    count = 0
+    replies = [None] * len(msg_ids)
+    while count < len(msg_ids):
+        get_msg = getattr(kc, f"get_{channel}_msg")
+        reply = get_msg(timeout=timeout)
+        try:
+            msg_id = reply["parent_header"]["msg_id"]
+            replies[msg_ids.index(msg_id)] = reply
+            count += 1
+        except ValueError:
+            # Allow debugging ignored replies
+            print(f"Ignoring reply not to any of {msg_ids}: {reply}")
+    return replies
+
+
 def execute(code="", kc=None, **kwargs):
     """wrapper for doing common steps for validating an execution request"""
     from .test_message_spec import validate_message