pythongh-119517: Fix several issues when pasting lot of text in the REPL

* Restore signal handlers for SIGINT and SIGSTOP (Ctrl-C and Ctrl-Z)
* Ensure that signals are processed as soon as possible by making
  reads more efficient.
* Protect against invalid state in internal REPL functions when
  interrumpted.
* Do not show extraneous newlines above the scroll buffer when pasting
  text in the REPL

Signed-off-by: Pablo Galindo <pablogsal@gmail.com>

Lib/_pyrepl/commands.py

@@ -482,3 +482,4 @@ def do(self) -> None:
  self.reader.in_bracketed_paste = False
  self.reader.dirty = True
  self.reader.calc_screen = self.reader.calc_complete_screen
+ self.reader.scroll_on_next_refresh = False

Lib/_pyrepl/console.py

@@ -69,7 +69,8 @@ def __init__(
  self.output_fd = f_out.fileno()
 
  @abstractmethod
- def refresh(self, screen: list[str], xy: tuple[int, int]) -> None: ...
+ def refresh(self, screen: list[str], xy: tuple[int, int],
+ scroll: bool = False) -> None: ...
 
  @abstractmethod
  def prepare(self) -> None: ...

Lib/_pyrepl/reader.py

@@ -240,6 +240,8 @@ class Reader:
  lxy: tuple[int, int] = field(init=False)
  calc_screen: CalcScreen = field(init=False)
  scheduled_commands: list[str] = field(default_factory=list)
+ can_colorize: bool = False
+ scroll_on_next_refresh: bool = True
 
  def __post_init__(self) -> None:
  # Enable the use of `insert` without a `prepare` call - necessary to
@@ -253,13 +255,16 @@ def __post_init__(self) -> None:
  self.cxy = self.pos2xy()
  self.lxy = (self.pos, 0)
  self.calc_screen = self.calc_complete_screen
-
+ self.can_colorize = can_colorize()
+
  def collect_keymap(self) -> tuple[tuple[KeySpec, CommandName], ...]:
  return default_keymap
 
  def append_to_screen(self) -> list[str]:
  new_screen = self.screen.copy() or ['']
 
+ if not self.buffer:
+ return []
  new_character = self.buffer[-1]
  new_character_len = wlen(new_character)
 
@@ -468,7 +473,7 @@ def get_prompt(self, lineno: int, cursor_on_line: bool) -> str:
  else:
  prompt = self.ps1
 
- if can_colorize():
+ if self.can_colorize:
  prompt = f"{ANSIColors.BOLD_MAGENTA}{prompt}{ANSIColors.RESET}"
  return prompt
 
@@ -606,8 +611,9 @@ def refresh(self) -> None:
  """Recalculate and refresh the screen."""
  # this call sets up self.cxy, so call it first.
  self.screen = self.calc_screen()
- self.console.refresh(self.screen, self.cxy)
+ self.console.refresh(self.screen, self.cxy, scroll=self.scroll_on_next_refresh)
  self.dirty = False
+ self.scroll_on_next_refresh = True
 
  def do_cmd(self, cmd: tuple[str, list[str]]) -> None:
  """`cmd` is a tuple of "event_name" and "event", which in the current

Lib/_pyrepl/unix_console.py

@@ -27,7 +27,6 @@
 import select
 import signal
 import struct
-import sys
 import termios
 import time
 from fcntl import ioctl
@@ -206,7 +205,7 @@ def change_encoding(self, encoding: str) -> None:
  """
  self.encoding = encoding
 
- def refresh(self, screen, c_xy):
+ def refresh(self, screen, c_xy, scroll=True):
  """
  Refresh the console screen.
 
@@ -248,22 +247,23 @@ def refresh(self, screen, c_xy):
  newscr = screen[offset : offset + height]
 
  # use hardware scrolling if we have it.
- if old_offset > offset and self._ri:
- self.__hide_cursor()
- self.__write_code(self._cup, 0, 0)
- self.__posxy = 0, old_offset
- for i in range(old_offset - offset):
- self.__write_code(self._ri)
- oldscr.pop(-1)
- oldscr.insert(0, "")
- elif old_offset < offset and self._ind:
- self.__hide_cursor()
- self.__write_code(self._cup, self.height - 1, 0)
- self.__posxy = 0, old_offset + self.height - 1
- for i in range(offset - old_offset):
- self.__write_code(self._ind)
- oldscr.pop(0)
- oldscr.append("")
+ if scroll:
+ if old_offset > offset and self._ri:
+ self.__hide_cursor()
+ self.__write_code(self._cup, 0, 0)
+ self.__posxy = 0, old_offset
+ for i in range(old_offset - offset):
+ self.__write_code(self._ri)
+ oldscr.pop(-1)
+ oldscr.insert(0, "")
+ elif old_offset < offset and self._ind:
+ self.__hide_cursor()
+ self.__write_code(self._cup, self.height - 1, 0)
+ self.__posxy = 0, old_offset + self.height - 1
+ for i in range(offset - old_offset):
+ self.__write_code(self._ind)
+ oldscr.pop(0)
+ oldscr.append("")
 
  self.__offset = offset
 
@@ -310,14 +310,13 @@ def prepare(self):
  """
  self.__svtermstate = tcgetattr(self.input_fd)
  raw = self.__svtermstate.copy()
- raw.iflag &= ~(termios.BRKINT | termios.INPCK | termios.ISTRIP | termios.IXON)
+ raw.iflag &= ~(termios.INPCK | termios.ISTRIP | termios.IXON)
  raw.oflag &= ~(termios.OPOST)
  raw.cflag &= ~(termios.CSIZE | termios.PARENB)
  raw.cflag |= termios.CS8
- raw.lflag &= ~(
- termios.ICANON | termios.ECHO | termios.IEXTEN | (termios.ISIG * 1)
- )
- raw.cc[termios.VMIN] = 1
+ raw.iflag |= termios.BRKINT
+ raw.lflag &= ~(termios.ICANON | termios.ECHO | termios.IEXTEN)
+ raw.lflag |= termios.ISIG
  raw.cc[termios.VTIME] = 0
  tcsetattr(self.input_fd, termios.TCSADRAIN, raw)
 
@@ -370,10 +369,21 @@ def get_event(self, block: bool = True) -> Event | None:
  Returns:
  - Event: Event object from the event queue.
  """
+ if self.wait(timeout=0):
+ try:
+ chars = os.read(self.input_fd, 1024)
+ for char in chars:
+ self.push_char(char)
+ except OSError as err:
+ if err.errno == errno.EINTR:
+ raise
+
  while self.event_queue.empty():
  while True:
  try:
- self.push_char(os.read(self.input_fd, 1))
+ chars = os.read(self.input_fd, 1024)
+ for char in chars:
+ self.push_char(char)
  except OSError as err:
  if err.errno == errno.EINTR:
  if not self.event_queue.empty():

Lib/_pyrepl/windows_console.py

@@ -136,7 +136,7 @@ def __init__(
  # Console I/O is redirected, fallback...
  self.out = None
 
- def refresh(self, screen: list[str], c_xy: tuple[int, int]) -> None:
+ def refresh(self, screen: list[str], c_xy: tuple[int, int], scroll: bool = True) -> None:
  """
  Refresh the console screen.
 
@@ -165,12 +165,13 @@ def refresh(self, screen: list[str], c_xy: tuple[int, int]) -> None:
  offset = cy - height + 1
  scroll_lines = offset - old_offset
 
- # Scrolling the buffer as the current input is greater than the visible
- # portion of the window. We need to scroll the visible portion and the
- # entire history
- self._scroll(scroll_lines, self._getscrollbacksize())
- self.__posxy = self.__posxy[0], self.__posxy[1] + scroll_lines
- self.__offset += scroll_lines
+ if scroll:
+ # Scrolling the buffer as the current input is greater than the visible
+ # portion of the window. We need to scroll the visible portion and the
+ # entire history
+ self._scroll(scroll_lines, self._getscrollbacksize())
+ self.__posxy = self.__posxy[0], self.__posxy[1] + scroll_lines
+ self.__offset += scroll_lines
 
  for i in range(scroll_lines):
  self.screen.append("")

Misc/NEWS.d/next/Core and Builtins/2024-06-07-16-42-17.gh-issue-119517.GXgQNl.rst

@@ -0,0 +1,2 @@
+Fix extraneous new lines in the scroll buffer when pasting in the REPL and
+make signals work again to interrupt slow operations. Patch by Pablo Galindo