test control channel priority

Author: minrk

ipykernel/tests/test_kernel.py

@@ -417,3 +417,48 @@ def test_interrupt_during_pdb_set_trace():
         # If we failed to interrupt interrupt, this will timeout:
         reply = get_reply(kc, msg_id2, TIMEOUT)
         validate_message(reply, 'execute_reply', msg_id2)
+
+
+def test_control_thread_priority():
+
+    N = 5
+    with new_kernel() as kc:
+        msg_id = kc.execute("pass")
+        get_reply(kc, msg_id)
+
+        sleep_msg_id = kc.execute("import asyncio; await asyncio.sleep(2)")
+
+        # submit N shell messages
+        shell_msg_ids = []
+        for i in range(N):
+            shell_msg_ids.append(kc.execute(f"i = {i}"))
+
+        # ensure all shell messages have arrived at the kernel before any control messages
+        time.sleep(0.5)
+        # at this point, shell messages should be waiting in msg_queue,
+        # rather than zmq while the kernel is still in the middle of processing
+        # the first execution
+
+        # now send N control messages
+        control_msg_ids = []
+        for i in range(N):
+            msg = kc.session.msg("kernel_info_request", {})
+            kc.control_channel.send(msg)
+            control_msg_ids.append(msg["header"]["msg_id"])
+
+        # finally, collect the replies on both channels for comparison
+        sleep_reply = get_reply(kc, sleep_msg_id)
+        shell_replies = []
+        for msg_id in shell_msg_ids:
+            shell_replies.append(get_reply(kc, msg_id))
+
+        control_replies = []
+        for msg_id in control_msg_ids:
+            control_replies.append(get_reply(kc, msg_id, channel="control"))
+
+    # verify that all control messages were handled before all shell messages
+    shell_dates = [msg["header"]["date"] for msg in shell_replies]
+    control_dates = [msg["header"]["date"] for msg in control_replies]
+    # comparing first to last ought to be enough, since queues preserve order
+    # use <= in case of very-fast handling and/or low resolution timers
+    assert control_dates[-1] <= shell_dates[0]

ipykernel/tests/utils.py

@@ -53,13 +53,15 @@ def flush_channels(kc=None):
                 validate_message(msg)
 
 
-def get_reply(kc, msg_id, timeout):
-    timeout = TIMEOUT
+def get_reply(kc, msg_id, timeout=TIMEOUT, channel='shell'):
     t0 = time()
     while True:
-        reply = kc.get_shell_msg(timeout=timeout)
+        get_msg = getattr(kc, f'get_{channel}_msg')
+        reply = get_msg(timeout=timeout)
         if reply['parent_header']['msg_id'] == msg_id:
             break
+        # Allow debugging ignored replies
+        print(f"Ignoring reply not to {msg_id}: {reply}")
         t1 = time()
         timeout -= t1 - t0
         t0 = t1