mirror of
https://github.com/python/cpython.git
synced 2025-07-25 20:24:11 +00:00

gh-119517: Fixes for pasting in pyrepl (GH-120253)
* Remove pyrepl's optimization for self-insert
This will be replaced by a less specialized optimization.
* Use line-buffering when pyrepl echoes pastes
Previously echoing was totally suppressed until the entire command had
been pasted and the terminal ended paste mode, but this gives the user
no feedback to indicate that an operation is in progress. Drawing
something to the screen once per line strikes a balance between
perceived responsiveness and performance.
* Remove dead code from pyrepl
`msg_at_bottom` is always true.
* Speed up pyrepl's screen rendering computation
The Reader in pyrepl doesn't hold a complete representation of the
screen area being drawn as persistent state. Instead, it recomputes it,
on each keypress. This is fast enough for a few hundred bytes, but
incredibly slow as the input buffer grows into the kilobytes (likely
because of pasting).
Rather than making some expensive and expansive changes to the repl's
internal representation of the screen, add some caching: remember some
data from one refresh to the next about what was drawn to the screen
and, if we don't find anything that has invalidated the results that
were computed last time around, reuse them. To keep this caching as
simple as possible, all we'll do is look for lines in the buffer that
were above the cursor the last time we were asked to update the screen,
and that are still above the cursor now. We assume that nothing can
affect a line that comes before both the old and new cursor location
without us being informed. Based on this assumption, we can reuse old
lines, which drastically speeds up the overwhelmingly common case where
the user is typing near the end of the buffer.
* Speed up pyrepl prompt drawing
Cache the `can_colorize()` call rather than repeatedly recomputing it.
This call looks up an environment variable, and is called once per
character typed at the REPL. The environment variable lookup shows up as
a hot spot when profiling, and we don't expect this to change while the
REPL is running.
* Speed up pasting multiple lines into the REPL
Previously, we were checking whether the command should be accepted each
time a line break was encountered, but that's not the expected behavior.
In bracketed paste mode, we expect everything pasted to be part of
a single block of code, and encountering a newline shouldn't behave like
a user pressing <Enter> to execute a command. The user should always
have a chance to review the pasted command before running it.
* Use a read buffer for input in pyrepl
Previously we were reading one byte at a time, which causes much slower
IO than necessary. Instead, read in chunks, processing previously read
data before asking for more.
* Optimize finding width of a single character
`wlen` finds the width of a multi-character string by adding up the
width of each character, and then subtracting the width of any escape
sequences. It's often called for single character strings, however,
which can't possibly contain escape sequences. Optimize for that case.
* Optimize disp_str for ASCII characters
Since every ASCII character is known to display as single width, we can
avoid not only the Unicode data lookup in `disp_str` but also the one
hidden in `str_width` for them.
* Speed up cursor movements in long pyrepl commands
When the current pyrepl command buffer contains many lines, scrolling up
becomes slow. We have optimizations in place to reuse lines above the
cursor position from one refresh to the next, but don't currently try to
reuse lines below the cursor position in the same way, so we wind up
with quadratic behavior where all lines of the buffer below the cursor
are recomputed each time the cursor moves up another line.
Optimize this by only computing one screen's worth of lines beyond the
cursor position. Any lines beyond that can't possibly be shown by the
console, and bounding this makes scrolling up have linear time
complexity instead.
---------
(cherry picked from commit 32a0faba43
)
Signed-off-by: Matt Wozniski <mwozniski@bloomberg.net>
Co-authored-by: Matt Wozniski <mwozniski@bloomberg.net>
Co-authored-by: Pablo Galindo <pablogsal@gmail.com>
483 lines
12 KiB
Python
483 lines
12 KiB
Python
# Copyright 2000-2010 Michael Hudson-Doyle <micahel@gmail.com>
|
|
# Antonio Cuni
|
|
# Armin Rigo
|
|
#
|
|
# All Rights Reserved
|
|
#
|
|
#
|
|
# Permission to use, copy, modify, and distribute this software and
|
|
# its documentation for any purpose is hereby granted without fee,
|
|
# provided that the above copyright notice appear in all copies and
|
|
# that both that copyright notice and this permission notice appear in
|
|
# supporting documentation.
|
|
#
|
|
# THE AUTHOR MICHAEL HUDSON DISCLAIMS ALL WARRANTIES WITH REGARD TO
|
|
# THIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
|
|
# AND FITNESS, IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL,
|
|
# INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
|
|
# RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
|
|
# CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
|
|
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
|
|
|
|
from __future__ import annotations
|
|
import os
|
|
|
|
# Categories of actions:
|
|
# killing
|
|
# yanking
|
|
# motion
|
|
# editing
|
|
# history
|
|
# finishing
|
|
# [completion]
|
|
|
|
|
|
# types
|
|
if False:
|
|
from .historical_reader import HistoricalReader
|
|
|
|
|
|
class Command:
|
|
finish: bool = False
|
|
kills_digit_arg: bool = True
|
|
|
|
def __init__(
|
|
self, reader: HistoricalReader, event_name: str, event: list[str]
|
|
) -> None:
|
|
# Reader should really be "any reader" but there's too much usage of
|
|
# HistoricalReader methods and fields in the code below for us to
|
|
# refactor at the moment.
|
|
|
|
self.reader = reader
|
|
self.event = event
|
|
self.event_name = event_name
|
|
|
|
def do(self) -> None:
|
|
pass
|
|
|
|
|
|
class KillCommand(Command):
|
|
def kill_range(self, start: int, end: int) -> None:
|
|
if start == end:
|
|
return
|
|
r = self.reader
|
|
b = r.buffer
|
|
text = b[start:end]
|
|
del b[start:end]
|
|
if is_kill(r.last_command):
|
|
if start < r.pos:
|
|
r.kill_ring[-1] = text + r.kill_ring[-1]
|
|
else:
|
|
r.kill_ring[-1] = r.kill_ring[-1] + text
|
|
else:
|
|
r.kill_ring.append(text)
|
|
r.pos = start
|
|
r.dirty = True
|
|
|
|
|
|
class YankCommand(Command):
|
|
pass
|
|
|
|
|
|
class MotionCommand(Command):
|
|
pass
|
|
|
|
|
|
class EditCommand(Command):
|
|
pass
|
|
|
|
|
|
class FinishCommand(Command):
|
|
finish = True
|
|
pass
|
|
|
|
|
|
def is_kill(command: type[Command] | None) -> bool:
|
|
return command is not None and issubclass(command, KillCommand)
|
|
|
|
|
|
def is_yank(command: type[Command] | None) -> bool:
|
|
return command is not None and issubclass(command, YankCommand)
|
|
|
|
|
|
# etc
|
|
|
|
|
|
class digit_arg(Command):
|
|
kills_digit_arg = False
|
|
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
c = self.event[-1]
|
|
if c == "-":
|
|
if r.arg is not None:
|
|
r.arg = -r.arg
|
|
else:
|
|
r.arg = -1
|
|
else:
|
|
d = int(c)
|
|
if r.arg is None:
|
|
r.arg = d
|
|
else:
|
|
if r.arg < 0:
|
|
r.arg = 10 * r.arg - d
|
|
else:
|
|
r.arg = 10 * r.arg + d
|
|
r.dirty = True
|
|
|
|
|
|
class clear_screen(Command):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
r.console.clear()
|
|
r.dirty = True
|
|
|
|
|
|
class refresh(Command):
|
|
def do(self) -> None:
|
|
self.reader.dirty = True
|
|
|
|
|
|
class repaint(Command):
|
|
def do(self) -> None:
|
|
self.reader.dirty = True
|
|
self.reader.console.repaint()
|
|
|
|
|
|
class kill_line(KillCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
b = r.buffer
|
|
eol = r.eol()
|
|
for c in b[r.pos : eol]:
|
|
if not c.isspace():
|
|
self.kill_range(r.pos, eol)
|
|
return
|
|
else:
|
|
self.kill_range(r.pos, eol + 1)
|
|
|
|
|
|
class unix_line_discard(KillCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
self.kill_range(r.bol(), r.pos)
|
|
|
|
|
|
class unix_word_rubout(KillCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
for i in range(r.get_arg()):
|
|
self.kill_range(r.bow(), r.pos)
|
|
|
|
|
|
class kill_word(KillCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
for i in range(r.get_arg()):
|
|
self.kill_range(r.pos, r.eow())
|
|
|
|
|
|
class backward_kill_word(KillCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
for i in range(r.get_arg()):
|
|
self.kill_range(r.bow(), r.pos)
|
|
|
|
|
|
class yank(YankCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
if not r.kill_ring:
|
|
r.error("nothing to yank")
|
|
return
|
|
r.insert(r.kill_ring[-1])
|
|
|
|
|
|
class yank_pop(YankCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
b = r.buffer
|
|
if not r.kill_ring:
|
|
r.error("nothing to yank")
|
|
return
|
|
if not is_yank(r.last_command):
|
|
r.error("previous command was not a yank")
|
|
return
|
|
repl = len(r.kill_ring[-1])
|
|
r.kill_ring.insert(0, r.kill_ring.pop())
|
|
t = r.kill_ring[-1]
|
|
b[r.pos - repl : r.pos] = t
|
|
r.pos = r.pos - repl + len(t)
|
|
r.dirty = True
|
|
|
|
|
|
class interrupt(FinishCommand):
|
|
def do(self) -> None:
|
|
import signal
|
|
|
|
self.reader.console.finish()
|
|
self.reader.finish()
|
|
os.kill(os.getpid(), signal.SIGINT)
|
|
|
|
|
|
class ctrl_c(Command):
|
|
def do(self) -> None:
|
|
self.reader.console.finish()
|
|
self.reader.finish()
|
|
raise KeyboardInterrupt
|
|
|
|
|
|
class suspend(Command):
|
|
def do(self) -> None:
|
|
import signal
|
|
|
|
r = self.reader
|
|
p = r.pos
|
|
r.console.finish()
|
|
os.kill(os.getpid(), signal.SIGSTOP)
|
|
## this should probably be done
|
|
## in a handler for SIGCONT?
|
|
r.console.prepare()
|
|
r.pos = p
|
|
# r.posxy = 0, 0 # XXX this is invalid
|
|
r.dirty = True
|
|
r.console.screen = []
|
|
|
|
|
|
class up(MotionCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
for _ in range(r.get_arg()):
|
|
x, y = r.pos2xy()
|
|
new_y = y - 1
|
|
|
|
if r.bol() == 0:
|
|
if r.historyi > 0:
|
|
r.select_item(r.historyi - 1)
|
|
return
|
|
r.pos = 0
|
|
r.error("start of buffer")
|
|
return
|
|
|
|
if (
|
|
x
|
|
> (
|
|
new_x := r.max_column(new_y)
|
|
) # we're past the end of the previous line
|
|
or x == r.max_column(y)
|
|
and any(
|
|
not i.isspace() for i in r.buffer[r.bol() :]
|
|
) # move between eols
|
|
):
|
|
x = new_x
|
|
|
|
r.setpos_from_xy(x, new_y)
|
|
|
|
|
|
class down(MotionCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
b = r.buffer
|
|
for _ in range(r.get_arg()):
|
|
x, y = r.pos2xy()
|
|
new_y = y + 1
|
|
|
|
if new_y > r.max_row():
|
|
if r.historyi < len(r.history):
|
|
r.select_item(r.historyi + 1)
|
|
r.pos = r.eol(0)
|
|
return
|
|
r.pos = len(b)
|
|
r.error("end of buffer")
|
|
return
|
|
|
|
if (
|
|
x
|
|
> (
|
|
new_x := r.max_column(new_y)
|
|
) # we're past the end of the previous line
|
|
or x == r.max_column(y)
|
|
and any(
|
|
not i.isspace() for i in r.buffer[r.bol() :]
|
|
) # move between eols
|
|
):
|
|
x = new_x
|
|
|
|
r.setpos_from_xy(x, new_y)
|
|
|
|
|
|
class left(MotionCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
for i in range(r.get_arg()):
|
|
p = r.pos - 1
|
|
if p >= 0:
|
|
r.pos = p
|
|
else:
|
|
self.reader.error("start of buffer")
|
|
|
|
|
|
class right(MotionCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
b = r.buffer
|
|
for i in range(r.get_arg()):
|
|
p = r.pos + 1
|
|
if p <= len(b):
|
|
r.pos = p
|
|
else:
|
|
self.reader.error("end of buffer")
|
|
|
|
|
|
class beginning_of_line(MotionCommand):
|
|
def do(self) -> None:
|
|
self.reader.pos = self.reader.bol()
|
|
|
|
|
|
class end_of_line(MotionCommand):
|
|
def do(self) -> None:
|
|
self.reader.pos = self.reader.eol()
|
|
|
|
|
|
class home(MotionCommand):
|
|
def do(self) -> None:
|
|
self.reader.pos = 0
|
|
|
|
|
|
class end(MotionCommand):
|
|
def do(self) -> None:
|
|
self.reader.pos = len(self.reader.buffer)
|
|
|
|
|
|
class forward_word(MotionCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
for i in range(r.get_arg()):
|
|
r.pos = r.eow()
|
|
|
|
|
|
class backward_word(MotionCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
for i in range(r.get_arg()):
|
|
r.pos = r.bow()
|
|
|
|
|
|
class self_insert(EditCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
text = self.event * r.get_arg()
|
|
r.insert(text)
|
|
|
|
|
|
class insert_nl(EditCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
r.insert("\n" * r.get_arg())
|
|
|
|
|
|
class transpose_characters(EditCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
b = r.buffer
|
|
s = r.pos - 1
|
|
if s < 0:
|
|
r.error("cannot transpose at start of buffer")
|
|
else:
|
|
if s == len(b):
|
|
s -= 1
|
|
t = min(s + r.get_arg(), len(b) - 1)
|
|
c = b[s]
|
|
del b[s]
|
|
b.insert(t, c)
|
|
r.pos = t
|
|
r.dirty = True
|
|
|
|
|
|
class backspace(EditCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
b = r.buffer
|
|
for i in range(r.get_arg()):
|
|
if r.pos > 0:
|
|
r.pos -= 1
|
|
del b[r.pos]
|
|
r.dirty = True
|
|
else:
|
|
self.reader.error("can't backspace at start")
|
|
|
|
|
|
class delete(EditCommand):
|
|
def do(self) -> None:
|
|
r = self.reader
|
|
b = r.buffer
|
|
if (
|
|
r.pos == 0
|
|
and len(b) == 0 # this is something of a hack
|
|
and self.event[-1] == "\004"
|
|
):
|
|
r.update_screen()
|
|
r.console.finish()
|
|
raise EOFError
|
|
for i in range(r.get_arg()):
|
|
if r.pos != len(b):
|
|
del b[r.pos]
|
|
r.dirty = True
|
|
else:
|
|
self.reader.error("end of buffer")
|
|
|
|
|
|
class accept(FinishCommand):
|
|
def do(self) -> None:
|
|
pass
|
|
|
|
|
|
class help(Command):
|
|
def do(self) -> None:
|
|
import _sitebuiltins
|
|
|
|
with self.reader.suspend():
|
|
self.reader.msg = _sitebuiltins._Helper()() # type: ignore[assignment, call-arg]
|
|
|
|
|
|
class invalid_key(Command):
|
|
def do(self) -> None:
|
|
pending = self.reader.console.getpending()
|
|
s = "".join(self.event) + pending.data
|
|
self.reader.error("`%r' not bound" % s)
|
|
|
|
|
|
class invalid_command(Command):
|
|
def do(self) -> None:
|
|
s = self.event_name
|
|
self.reader.error("command `%s' not known" % s)
|
|
|
|
|
|
class show_history(Command):
|
|
def do(self) -> None:
|
|
from .pager import get_pager
|
|
from site import gethistoryfile # type: ignore[attr-defined]
|
|
|
|
history = os.linesep.join(self.reader.history[:])
|
|
with self.reader.suspend():
|
|
pager = get_pager()
|
|
pager(history, gethistoryfile())
|
|
|
|
|
|
class paste_mode(Command):
|
|
|
|
def do(self) -> None:
|
|
self.reader.paste_mode = not self.reader.paste_mode
|
|
self.reader.dirty = True
|
|
|
|
|
|
class enable_bracketed_paste(Command):
|
|
def do(self) -> None:
|
|
self.reader.paste_mode = True
|
|
self.reader.in_bracketed_paste = True
|
|
|
|
class disable_bracketed_paste(Command):
|
|
def do(self) -> None:
|
|
self.reader.paste_mode = False
|
|
self.reader.in_bracketed_paste = False
|
|
self.reader.dirty = True
|