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

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,22 @@ def get_event(self, block: bool = True) -> Event | None:
         Returns:
         - Event: Event object from the event queue.
         """
+        BUFFER_SIZE = 1024*100
+        if self.wait(timeout=0):
+            try:
+                chars = os.read(self.input_fd, BUFFER_SIZE)
+                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, BUFFER_SIZE)
+                    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("")

Lib/test/test_pyrepl/support.py

@@ -103,7 +103,7 @@ def getpending(self) -> Event:
     def getheightwidth(self) -> tuple[int, int]:
         return self.height, self.width
 
-    def refresh(self, screen: list[str], xy: tuple[int, int]) -> None:
+    def refresh(self, screen: list[str], xy: tuple[int, int], scroll: bool = True) -> None:
         pass
 
     def prepare(self) -> None:

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