Code to run the benchmark kitten headlessly

benchmark.py

@@ -0,0 +1,90 @@
+#!./kitty/launcher/kitty +launch
+# License: GPL v3 Copyright: 2016, Kovid Goyal <kovid at kovidgoyal.net>
+
+import fcntl
+import io
+import os
+import select
+import signal
+import struct
+import sys
+import termios
+import time
+from pty import CHILD, fork
+
+from kitty.constants import kitten_exe
+from kitty.fast_data_types import Screen
+from kitty.utils import read_screen_size
+
+
+def run_parsing_benchmark(cell_width: int = 10, cell_height: int = 20, scrollback: int = 20000) -> None:
+    isatty = sys.stdout.isatty()
+    if isatty:
+        sz = read_screen_size()
+        columns, rows = sz.cols, sz.rows
+    else:
+        columns, rows = 80, 25
+    child_pid, master_fd = fork()
+    is_child = child_pid == CHILD
+    argv = [kitten_exe(), '__benchmark__', '--with-scrollback']
+    if is_child:
+        while read_screen_size().width != columns * cell_width:
+            time.sleep(0.01)
+        signal.pthread_sigmask(signal.SIG_SETMASK, ())
+        os.execvp(argv[0], argv)
+    # os.set_blocking(master_fd, False)
+    x_pixels = columns * cell_width
+    y_pixels = rows * cell_height
+    s = struct.pack('HHHH', rows, columns, x_pixels, y_pixels)
+    fcntl.ioctl(master_fd, termios.TIOCSWINSZ, s)
+
+    write_buf = b''
+    r_pipe, w_pipe = os.pipe2(os.O_CLOEXEC | os.O_NONBLOCK)
+    class ToChild:
+        def write(self, x: bytes | str) -> None:
+            nonlocal write_buf
+            if isinstance(x, str):
+                x = x.encode()
+            write_buf += x
+            os.write(w_pipe, b'1')
+
+    screen = Screen(None, rows, columns, scrollback, cell_width, cell_height, 0, ToChild())
+
+    def parse_bytes(data: bytes) -> None:
+        data = memoryview(data)
+        while data:
+            dest = screen.test_create_write_buffer()
+            s = screen.test_commit_write_buffer(data, dest)
+            data = data[s:]
+            screen.test_parse_written_data()
+
+
+    while True:
+        rd, wd, _ = select.select([master_fd, r_pipe], [master_fd] if write_buf else [], [])
+        if r_pipe in rd:
+            os.read(r_pipe, 256)
+        if master_fd in rd:
+            try:
+                data = os.read(master_fd, io.DEFAULT_BUFFER_SIZE)
+            except OSError:
+                data = b''
+            if not data:
+                break
+            parse_bytes(data)
+        if master_fd in wd:
+            n = os.write(master_fd, write_buf)
+            write_buf = write_buf[n:]
+    if isatty:
+        lines: list[str] = []
+        screen.linebuf.as_ansi(lines.append)
+        sys.stdout.write(''.join(lines))
+    else:
+        sys.stdout.write(str(screen.linebuf))
+
+
+def main() -> None:
+    run_parsing_benchmark()
+
+
+if __name__ == '__main__':
+    main()

kitty/fast_data_types.pyi

@@ -1164,6 +1164,8 @@ class LineBuf:
     def line(self, num: int) -> Line:
         pass
 
+    def as_ansi(self, callback: Callable[[str], None]) -> None: ...
+
 
 class Cursor:
     x: int
@@ -1205,6 +1207,10 @@ class Screen:
     ):
         pass
 
+    def test_create_write_buffer(self) -> memoryview: ...
+    def test_commit_write_buffer(self, inp: memoryview, output: memoryview) -> int: ...
+    def test_parse_written_data(self, dump_callback: None = None) -> None: ...
+
     def cursor_at_prompt(self) -> bool:
         pass