Only allow KeyboardInterrupt at specific times

Because our goal is to proxy interrupt signals to the remote process, it's important that we only accept those signals at moments when we're ready to handle them and send them on. In particular, we're prepared to handle them when reading user input for a prompt, and when waiting for the server to send a new prompt after our last command. We do not want to accept them at other times, like while we're in the middle of printing output to the screen, as otherwise a while True: "Hello" executed at the PDB REPL can't reliably be Ctrl-C'd: it's more likely that the signal will arrive while the client is printing than while it's waiting for the next line from the server, and if so nothing will be prepared to handle the `KeyboardInterrupt` and the client will exit.
python · pablogsal · May 5, 2025 · Apr 30, 2025 · Apr 30, 2025 · Apr 30, 2025
commit 8d89bf4bfd57bca5e86eed541f4bce998668ae81
@@ -3009,12 +3009,53 @@ def readline_completion(self, completer):
         finally:
             readline.set_completer(old_completer)
 
+    if not hasattr(signal, "pthread_sigmask"):
+        # On Windows, we must drop signals arriving while we're ignoring them.
+        @contextmanager
+        def _block_sigint(self):
+            old_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
+            try:
+                yield
+            finally:
+                signal.signal(signal.SIGINT, old_handler)
+
+        @contextmanager
+        def _handle_sigint(self, handler):
+            old_handler = signal.signal(signal.SIGINT, handler)
+            try:
+                yield
+            finally:
+                signal.signal(signal.SIGINT, old_handler)
+    else:
+        # On Unix, we can save them to be processed later.
+        @contextmanager
+        def _block_sigint(self):
+            signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT})
+            try:
+                yield
+            finally:
+                signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGINT})
+
+        @contextmanager
+        def _handle_sigint(self, handler):
+            old_handler = signal.signal(signal.SIGINT, handler)
+            try:
+                signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGINT})
+                yield
+            finally:
+                signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT})
+                signal.signal(signal.SIGINT, old_handler)
+
     def cmdloop(self):
-        with self.readline_completion(self.complete):
+        with (
+            self._block_sigint(),
+            self.readline_completion(self.complete),
+        ):
             while not self.write_failed:
                 try:
-                    if not (payload_bytes := self.sockfile.readline()):
-                        break
+                    with self._handle_sigint(signal.default_int_handler):
+                        if not (payload_bytes := self.sockfile.readline()):
+                            break
                 except KeyboardInterrupt:
                     self.send_interrupt()
                     continue
@@ -3061,7 +3102,8 @@ def process_payload(self, payload):
     def prompt_for_reply(self, prompt):
         while True:
             try:
-                payload = {"reply": self.read_command(prompt)}
+                with self._handle_sigint(signal.default_int_handler):
+                    payload = {"reply": self.read_command(prompt)}
             except EOFError:
                 payload = {"signal": "EOF"}
             except KeyboardInterrupt:
@@ -3101,7 +3143,8 @@ def complete(self, text, state):
             if self.write_failed:
                 return None
 
-            payload = self.sockfile.readline()
+            with self._handle_sigint(signal.default_int_handler):
+                payload = self.sockfile.readline()
             if not payload:
                 return None