0
0
mirror of https://github.com/python/cpython.git synced 2024-11-24 00:38:00 +01:00
cpython/Lib/_pyrepl/commands.py
Matt Wozniski 32a0faba43
gh-119517: Fixes for pasting in pyrepl (#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.

---------

Signed-off-by: Matt Wozniski <mwozniski@bloomberg.net>
Co-authored-by: Pablo Galindo <pablogsal@gmail.com>
2024-06-11 16:42:10 +00:00

484 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