diff --git a/kitty/child.py b/kitty/child.py index c5b85693d..16a0599b1 100644 --- a/kitty/child.py +++ b/kitty/child.py @@ -26,6 +26,8 @@ if is_macos: from kitty.fast_data_types import cmdline_of_process as cmdline_ from kitty.fast_data_types import cwd_of_process as _cwd from kitty.fast_data_types import environ_of_process as _environ_of_process + from kitty.fast_data_types import memory_of_process as _memory_of_process + from kitty.fast_data_types import ppid_of_process as _ppid_of_process from kitty.fast_data_types import process_group_map as _process_group_map def cwd_of_process(pid: int) -> str: @@ -44,6 +46,30 @@ if is_macos: def cmdline_of_pid(pid: int) -> list[str]: return cmdline_(pid) + + def _get_descendants_of_macos(pid: int) -> set[int]: + children_map: DefaultDict[int, list[int]] = defaultdict(list) + for p in fast_data_types.get_all_processes(): + with suppress(Exception): + children_map[_ppid_of_process(p)].append(p) + result: set[int] = set() + stack = list(children_map.get(pid, [])) + while stack: + child = stack.pop() + if child not in result: + result.add(child) + stack.extend(children_map.get(child, [])) + return result + + def memory_used_by_process_tree_rooted_at(pid: int, check_if_cgroup_root: bool = False) -> int: + with suppress(Exception): + pids = _get_descendants_of_macos(pid) + total = _memory_of_process(pid) # raises if pid doesn't exist + for p in pids: + with suppress(Exception): + total += _memory_of_process(p) + return total + return -1 else: def cmdline_of_pid(pid: int) -> list[str]: @@ -94,6 +120,63 @@ else: def abspath_of_exe(pid: int) -> str: return os.path.realpath(f'/proc/{pid}/exe', strict=True) + def _get_descendants_of(pid: int) -> set[int]: + result: set[int] = set() + stack = [pid] + while stack: + current = stack.pop() + with suppress(OSError): + with open(f'/proc/{current}/task/{current}/children') as f: + for child_str in f.read().split(): + child = int(child_str) + if child not in result: + result.add(child) + stack.append(child) + return result + + def _memory_from_smaps_rollup(pid: int) -> int: + with open(f'/proc/{pid}/smaps_rollup') as f: + for line in f: + if line.startswith('Pss:'): + return int(line.split()[1]) * 1024 + return 0 + + def memory_used_by_process_tree_rooted_at(pid: int, check_if_cgroup_root: bool = False) -> int: + with suppress(Exception): + with open(f'/proc/{pid}/cgroup') as f: + cgroup_line = f.readline().strip() + cgroup_path = cgroup_line.split(':')[2].lstrip('/') + cgroup_dir = os.path.join('/sys/fs/cgroup', cgroup_path) + + use_cgroup = True + if check_if_cgroup_root: + with suppress(OSError): + with open(os.path.join(cgroup_dir, 'cgroup.procs')) as f: + cgroup_pids = {int(x) for x in f.read().split() if x} + descendants = _get_descendants_of(pid) + descendants.add(pid) + use_cgroup = cgroup_pids <= descendants + + if use_cgroup: + target_keys = {'anon', 'shmem', 'kernel', 'sock', 'zswap'} + mem_bytes = 0 + with open(os.path.join(cgroup_dir, 'memory.stat')) as f: + for line in f: + parts = line.split() + if parts[0] in target_keys: + mem_bytes += int(parts[1]) + return mem_bytes + + # cgroup contains processes outside our tree; sum PSS per process + descendants = _get_descendants_of(pid) + descendants.add(pid) + mem_bytes = 0 + for p in descendants: + with suppress(OSError): + mem_bytes += _memory_from_smaps_rollup(p) + return mem_bytes + return -1 + @run_once def checked_terminfo_dir() -> str | None: @@ -619,3 +702,8 @@ class Child: termios.tcsetattr(self.child_fd, when, self.initial_termios_state) except OSError: pass + + def get_memory_used_by_child(self, check_if_cgroup_root: bool = False) -> int: + if self.pid is None: + return -1 + return memory_used_by_process_tree_rooted_at(self.pid, check_if_cgroup_root) diff --git a/kitty/fast_data_types.pyi b/kitty/fast_data_types.pyi index e883f8261..70b85c71b 100644 --- a/kitty/fast_data_types.pyi +++ b/kitty/fast_data_types.pyi @@ -335,7 +335,6 @@ LEFT_EDGE: int RIGHT_EDGE: int # }}} - def encode_key_for_tty( key: int = 0, shifted_key: int = 0, @@ -343,66 +342,50 @@ def encode_key_for_tty( mods: int = 0, action: int = 1, key_encoding_flags: int = 0, - text: str = "", - cursor_key_mode: bool = False + text: str = '', + cursor_key_mode: bool = False, ) -> str: pass - def log_error_string(s: str) -> None: pass - def glfw_get_key_name(key: int, native_key: int) -> Optional[str]: pass - StartupCtx = NewType('StartupCtx', int) Display = NewType('Display', int) - -def init_x11_startup_notification( - display: Display, - window_id: int, - startup_id: Optional[str] = None -) -> StartupCtx: +def init_x11_startup_notification(display: Display, window_id: int, startup_id: Optional[str] = None) -> StartupCtx: pass - def end_x11_startup_notification(ctx: StartupCtx) -> None: pass - def x11_display() -> Optional[Display]: pass - def user_cache_dir() -> str: pass - def process_group_map() -> Tuple[Tuple[int, int], ...]: pass - def environ_of_process(pid: int) -> str: pass - +def memory_of_process(pid: int) -> int: ... +def ppid_of_process(pid: int) -> int: ... def cmdline_of_process(pid: int) -> List[str]: pass - def cwd_of_process(pid: int) -> str: pass - def abspath_of_process(pid: int) -> str: ... - def default_color_table() -> Tuple[int, ...]: pass - class FontConfigPattern(TypedDict): descriptor_type: Literal['fontconfig'] path: str @@ -432,40 +415,32 @@ class FontConfigPattern(TypedDict): axes: NotRequired[Tuple[float, ...]] features: NotRequired[Tuple[ParsedFontFeature, ...]] - def fc_list(spacing: int = -1, allow_bitmapped_fonts: bool = False, only_variable: bool = False) -> Tuple[FontConfigPattern, ...]: pass - def fc_match( family: Optional[str] = None, bold: bool = False, italic: bool = False, spacing: int = FC_MONO, allow_bitmapped_fonts: bool = False, - size_in_pts: float = 0., - dpi: float = 0. + size_in_pts: float = 0.0, + dpi: float = 0.0, ) -> FontConfigPattern: pass - -def fc_match_postscript_name( - postscript_name: str -) -> FontConfigPattern: +def fc_match_postscript_name(postscript_name: str) -> FontConfigPattern: pass - def add_font_file(path: str) -> bool: ... def set_builtin_nerd_font(path: str) -> Union[CoreTextFont, FontConfigPattern]: ... - class FeatureData(TypedDict): name: NotRequired[str] tooltip: NotRequired[str] sample: NotRequired[str] params: NotRequired[Tuple[str, ...]] - class Face: path: Optional[str] def __init__(self, descriptor: Optional[FontConfigPattern] = None, path: str = '', index: int = 0): ... @@ -473,15 +448,12 @@ class Face: def identify_for_debug(self) -> str: ... def postscript_name(self) -> str: ... def set_size(self, sz_in_pts: float, dpi_x: float, dpi_y: float) -> None: ... - def render_sample_text( - self, text: str, width: int, height: int, fg_color: int = 0xffffff - ) -> tuple[bytes, int, int]: ... - def render_codepoint(self, cp: int, fg_color: int = 0xffffff) -> tuple[bytes, int, int]: ... + def render_sample_text(self, text: str, width: int, height: int, fg_color: int = 0xFFFFFF) -> tuple[bytes, int, int]: ... + def render_codepoint(self, cp: int, fg_color: int = 0xFFFFFF) -> tuple[bytes, int, int]: ... def get_variation(self) -> Optional[Dict[str, float]]: ... def get_features(self) -> Dict[str, Optional[FeatureData]]: ... def applied_features(self) -> Dict[str, str]: ... - class CoreTextFont(TypedDict): descriptor_type: Literal['core_text'] path: str @@ -505,7 +477,6 @@ class CoreTextFont(TypedDict): axis_map: NotRequired[Dict[str, float]] features: NotRequired[Tuple[ParsedFontFeature, ...]] - class CTFace: path: Optional[str] def __init__(self, descriptor: Optional[CoreTextFont] = None, path: str = ''): ... @@ -514,68 +485,52 @@ class CTFace: def postscript_name(self) -> str: ... def set_size(self, sz_in_pts: float, dpi_x: float, dpi_y: float) -> None: ... def render_sample_text( - self, text: str, width: int, height: int, fg_color: int = 0xffffff, + self, + text: str, + width: int, + height: int, + fg_color: int = 0xFFFFFF, ) -> tuple[bytes, int, int]: ... - def render_codepoint(self, cp: int, fg_color: int = 0xffffff) -> tuple[bytes, int, int]: ... + def render_codepoint(self, cp: int, fg_color: int = 0xFFFFFF) -> tuple[bytes, int, int]: ... def get_variation(self) -> Optional[Dict[str, float]]: ... def get_features(self) -> Dict[str, Optional[FeatureData]]: ... def applied_features(self) -> Dict[str, str]: ... - def coretext_all_fonts(monospaced_only: bool) -> Tuple[CoreTextFont, ...]: pass - class ParsedFontFeature: def __init__(self, s: str): ... - -def add_timer( - callback: Callable[[Optional[int]], None], - interval: float, - repeats: bool = True -) -> int: +def add_timer(callback: Callable[[Optional[int]], None], interval: float, repeats: bool = True) -> int: pass - def remove_timer(timer_id: int) -> None: pass - def monitor_pid(pid: int) -> None: pass - def add_window(os_window_id: int, tab_id: int, title: str) -> int: pass - -def compile_program( - which: int, vertex_shaders: Tuple[str, ...], fragment_shaders: Tuple[str, ...], allow_recompile: bool = False -) -> int: +def compile_program(which: int, vertex_shaders: Tuple[str, ...], fragment_shaders: Tuple[str, ...], allow_recompile: bool = False) -> int: pass - def init_cell_program() -> None: pass - def set_os_window_chrome(os_window_id: int) -> bool: pass - def set_borders_rects(os_window_id: int, tab_id: int, rects: list[Border]) -> None: ... - - def init_borders_program() -> None: pass def os_window_has_background_image(os_window_id: int) -> bool: pass - def dbus_set_notification_callback(c: Optional[Callable[[str, int, Union[str, int]], None]]) -> None: ... - def dbus_send_notification( app_name: str, app_icon: str, @@ -590,10 +545,7 @@ def dbus_send_notification( ) -> int: pass - def dbus_close_notification(dbus_notification_id: int) -> bool: ... - - def cocoa_send_notification( appname: str, identifier: str, @@ -610,10 +562,8 @@ def cocoa_send_notification( def cocoa_bundle_image_as_png(path_or_identifier: str, output_path: str = '', image_size: int = 256, image_type: int = 1) -> bytes: ... def cocoa_remove_delivered_notification(identifier: str) -> bool: ... def cocoa_live_delivered_notifications() -> bool: ... - def create_os_window( - get_window_size: Callable[[int, int, int, int, float, float], Tuple[int, - int]], + get_window_size: Callable[[int, int, int, int, float, float], Tuple[int, int]], pre_show_callback: Callable[[int], None], title: str, wm_class_name: str, @@ -627,99 +577,62 @@ def create_os_window( ) -> int: pass - -def update_window_title( - os_window_id: int, tab_id: int, window_id: int, title: str -) -> None: +def update_window_title(os_window_id: int, tab_id: int, window_id: int, title: str) -> None: pass - -def update_window_visibility( - os_window_id: int, tab_id: int, window_id: int, - visible: bool -) -> None: +def update_window_visibility(os_window_id: int, tab_id: int, window_id: int, visible: bool) -> None: pass - def sync_os_window_title(os_window_id: int) -> None: pass - -def set_options( - opts: Optional[Options], - is_wayland: bool = False, - debug_rendering: bool = False, - debug_font_fallback: bool = False -) -> None: +def set_options(opts: Optional[Options], is_wayland: bool = False, debug_rendering: bool = False, debug_font_fallback: bool = False) -> None: pass - def get_options() -> Options: pass - def glfw_primary_monitor_size() -> tuple[int, int]: pass def glfw_get_monitor_workarea() -> tuple[tuple[int, int, int, int], ...]: pass - def set_default_window_icon(path: str) -> None: pass - def set_os_window_icon(os_window_id: int, path: str | None | bytes = None) -> None: ... - - -def set_custom_cursor( - cursor_shape: str, - images: Tuple[Tuple[bytes, int, int], ...], - x: int = 0, - y: int = 0 -) -> None: +def set_custom_cursor(cursor_shape: str, images: Tuple[Tuple[bytes, int, int], ...], x: int = 0, y: int = 0) -> None: pass - def is_css_pointer_name_valid(name: str) -> bool: ... def pointer_name_to_css_name(name: str) -> str: ... - - def load_png_data(data: bytes) -> Tuple[bytes, int, int]: pass - def glfw_terminate() -> None: pass - def glfw_init( - path: str, edge_spacing_func: Callable[[EdgeLiteral], float], debug_keyboard: bool = False, debug_rendering: bool = False, - wayland_enable_ime: bool = True + path: str, edge_spacing_func: Callable[[EdgeLiteral], float], debug_keyboard: bool = False, debug_rendering: bool = False, wayland_enable_ime: bool = True ) -> tuple[bool, bool]: pass - def free_font_data() -> None: pass - def toggle_maximized(os_window_id: int = 0) -> bool: pass - def toggle_fullscreen(os_window_id: int = 0) -> bool: pass - def thread_write(fd: int, data: bytes) -> None: pass - def set_ignore_os_keyboard_processing(yes: bool) -> None: pass - def set_background_image( path: Optional[str], os_window_ids: Tuple[int, ...], @@ -734,27 +647,21 @@ def set_background_image( ) -> None: pass - def set_boss(boss: Boss) -> None: pass - def get_boss() -> Boss: # this can return None but we ignore that for convenience pass - def safe_pipe(nonblock: bool = True) -> Tuple[int, int]: pass - def patch_global_colors(spec: Dict[str, Optional[int]], configured: bool) -> None: pass - class Color: @classmethod def parse_color(cls, spec: str) -> Color | None: ... - @property def rgb(self) -> int: pass @@ -816,51 +723,40 @@ class Color: def contrast(self, other: 'Color') -> float: pass - class ColorProfile: - # The dynamic color properties return the current color value. Use delattr # to reset them to configured value. @property def default_fg(self) -> Color: ... @default_fg.setter - def default_fg(self, val: Union[int|Color]) -> None: ... - + def default_fg(self, val: Union[int | Color]) -> None: ... @property def default_bg(self) -> Color: ... @default_bg.setter - def default_bg(self, val: Union[int|Color]) -> None: ... - + def default_bg(self, val: Union[int | Color]) -> None: ... @property def cursor_color(self) -> Optional[Color]: ... @cursor_color.setter - def cursor_color(self, val: Union[None|int|Color]) -> None: ... - + def cursor_color(self, val: Union[None | int | Color]) -> None: ... @property def cursor_text_color(self) -> Optional[Color]: ... @cursor_text_color.setter - def cursor_text_color(self, val: Union[None|int|Color]) -> None: ... - + def cursor_text_color(self, val: Union[None | int | Color]) -> None: ... @property def highlight_fg(self) -> Optional[Color]: ... @highlight_fg.setter - def highlight_fg(self, val: Union[None|int|Color]) -> None: ... - + def highlight_fg(self, val: Union[None | int | Color]) -> None: ... @property def highlight_bg(self) -> Optional[Color]: ... @highlight_bg.setter - def highlight_bg(self, val: Union[None|int|Color]) -> None: ... - + def highlight_bg(self, val: Union[None | int | Color]) -> None: ... @property def visual_bell_color(self) -> Optional[Color]: ... @visual_bell_color.setter - def visual_bell_color(self, val: Union[None|int|Color]) -> None: ... - + def visual_bell_color(self, val: Union[None | int | Color]) -> None: ... def __init__(self, opts: Optional[Options] = None): ... - def as_dict(self) -> Dict[str, int | None | tuple[tuple[Color, float], ...]]: ... def basic_colors(self) -> Dict[str, int | None | tuple[tuple[Color, float], ...]]: ... - def as_color(self, val: int) -> Optional[Color]: pass @@ -875,178 +771,130 @@ class ColorProfile: def reload_from_opts(self, opts: Optional[Options] = None, reset_overriden_colors: bool = True) -> None: ... def palette_color_is_generated(self, idx: int) -> None: ... - def get_transparent_background_color(self, index: int) -> Color | None: ... def set_transparent_background_color(self, index: int, color: Color | None = None, opacity: float | None = None) -> None: ... - def patch_color_profiles( - spec: Dict[str, Optional[int]], transparent_background_colors: tuple[tuple[Color, float], ...], - profiles: Tuple[ColorProfile, ...], change_configured: bool + spec: Dict[str, Optional[int]], transparent_background_colors: tuple[tuple[Color, float], ...], profiles: Tuple[ColorProfile, ...], change_configured: bool ) -> None: pass - def create_canvas(d: bytes, w: int, x: int, y: int, cw: int, ch: int, bpp: int) -> bytes: ... - - -def os_window_font_size( - os_window_id: int, new_sz: float = -1., force: bool = False -) -> float: +def os_window_font_size(os_window_id: int, new_sz: float = -1.0, force: bool = False) -> float: pass - def cocoa_set_notification_activated_callback(identifier: Optional[Callable[[str, str, str], None]]) -> None: pass - def cocoa_set_global_shortcut(name: str, mods: int, key: int) -> bool: pass - def cocoa_get_lang() -> Tuple[str, str, str]: pass - def cocoa_set_url_handler(url_scheme: str, bundle_id: Optional[str] = None) -> None: pass - def cocoa_set_app_icon(icon_path: str, app_path: Optional[str] = None) -> None: pass - def cocoa_set_dock_icon(icon_path: str) -> None: pass - def cocoa_show_progress_bar_on_dock_icon(progress: float = -100) -> None: pass - def cocoa_hide_app() -> None: pass - def cocoa_hide_other_apps() -> None: pass - def cocoa_minimize_os_window(os_window_id: Optional[int] = None) -> None: pass - def locale_is_valid(name: str) -> bool: pass - def mark_os_window_for_close(os_window_id: int, cr_type: int = 2) -> bool: pass - def set_application_quit_request(cr_type: int = 2) -> None: pass - def current_application_quit_request() -> int: pass - -def global_font_size(val: float = -1.) -> float: +def global_font_size(val: float = -1.0) -> float: pass - def focus_os_window(os_window_id: int, also_raise: bool = True, activation_token: Optional[str] = None) -> bool: pass - def toggle_secure_input() -> None: pass - def macos_cycle_through_os_windows(backwards: bool) -> None: pass - def start_profiler(path: str) -> None: pass - def stop_profiler() -> None: pass - def destroy_global_data() -> None: pass - def current_os_window() -> Optional[int]: pass - def last_focused_os_window_id() -> int: pass - def current_focused_os_window_id() -> int: pass - def cocoa_set_menubar_title(title: str) -> None: pass - def change_os_window_state(state: int, os_window_id: Optional[int] = 0) -> None: pass - def change_background_opacity(os_window_id: int, opacity: float) -> bool: pass - def background_opacity_of(os_window_id: int) -> Optional[float]: pass - def read_command_response(fd: int, timeout: float, list: List[bytes]) -> None: pass - def wcswidth(string: str) -> int: pass - def is_emoji_presentation_base(code: int) -> bool: pass - def x11_window_id(os_window_id: int) -> int: pass - def cocoa_window_id(os_window_id: int) -> int: pass - def swap_tabs(os_window_id: int, a: int, b: int) -> None: ... def reorder_tabs(os_window_id: int, *tab_ids: int) -> None: ... - def set_active_tab(os_window_id: int, a: int) -> None: pass - def set_active_window(os_window_id: int, tab_id: int, window_id: int) -> None: pass - def ring_bell(os_window_id: int = 0) -> None: ... def request_attention(os_window_id: int) -> None: ... - - def concat_cells(cell_width: int, cell_height: int, is_32_bit: bool, cells: Tuple[bytes, ...], bgcolor: int = 0) -> bytes: pass - FontFace = Union[Face, CTFace] class CurrentFonts(TypedDict): @@ -1060,53 +908,38 @@ class CurrentFonts(TypedDict): logical_dpi_x: float logical_dpi_y: float - def current_fonts(os_window_id: int = 0) -> CurrentFonts: ... - - def remove_window(os_window_id: int, tab_id: int, window_id: int) -> None: pass - def remove_tab(os_window_id: int, tab_id: int) -> None: pass - def pt_to_px(pt: float, os_window_id: int = 0) -> int: pass - def next_window_id() -> int: pass - def mark_tab_bar_dirty(os_window_id: int, should_be_shown: bool) -> None: pass - def is_tab_bar_visible(os_window_id: int) -> bool: ... - - def detach_window(os_window_id: int, tab_id: int, window_id: int) -> None: pass - def attach_window(os_window_id: int, tab_id: int, window_id: int) -> None: pass - def add_tab(os_window_id: int) -> int: pass - def cell_size_for_window(os_window_id: int) -> Tuple[int, int]: pass - def wakeup_main_loop() -> None: pass - class Region: left: int top: int @@ -1118,93 +951,75 @@ class Region: def __init__(self, x: Tuple[int, int, int, int, int, int]): pass - -def viewport_for_window( - os_window_id: int -) -> Tuple[Region, Region, int, int, int, int]: +def viewport_for_window(os_window_id: int) -> Tuple[Region, Region, int, int, int, int]: pass - TermiosPtr = NewType('TermiosPtr', int) - def raw_tty(fd: int, termios_ptr: TermiosPtr, optional_actions: int = termios.TCSAFLUSH) -> None: pass - def close_tty(fd: int, termios_ptr: TermiosPtr, optional_actions: int = termios.TCSAFLUSH) -> None: pass - def normal_tty(fd: int, termios_ptr: TermiosPtr, optional_actions: int = termios.TCSAFLUSH) -> None: pass - def open_tty(read_with_timeout: bool = False, optional_actions: int = termios.TCSAFLUSH) -> Tuple[int, TermiosPtr]: pass - def parse_input_from_terminal( - text_callback: Callable[[str], None], dcs_callback: Callable[[str], None], - csi_callback: Callable[[str], None], osc_callback: Callable[[str], None], - pm_callback: Callable[[str], None], apc_callback: Callable[[str], None], - data: str, in_bracketed_paste: bool + text_callback: Callable[[str], None], + dcs_callback: Callable[[str], None], + csi_callback: Callable[[str], None], + osc_callback: Callable[[str], None], + pm_callback: Callable[[str], None], + apc_callback: Callable[[str], None], + data: str, + in_bracketed_paste: bool, ) -> str: pass - class Line: - def sprite_at(self, cell: int) -> int: ... - -def test_shape(line: Line, - path: Optional[str] = None, - index: int = 0) -> List[Tuple[int, int, int, Tuple[int, ...]]]: +def test_shape(line: Line, path: Optional[str] = None, index: int = 0) -> List[Tuple[int, int, int, Tuple[int, ...]]]: pass - def test_render_line(line: Line) -> None: pass - def sprite_map_set_limits(w: int, h: int) -> None: pass - -def set_send_sprite_to_gpu( - func: Optional[Callable[[int, int, int, bytes], None]] -) -> None: +def set_send_sprite_to_gpu(func: Optional[Callable[[int, int, int, bytes], None]]) -> None: pass - def set_font_data( - descriptor_for_idx: Callable[[int], Tuple[Union[FontObject|str], bool, bool]], - bold: int, italic: int, bold_italic: int, num_symbol_fonts: int, - symbol_maps: Tuple[Tuple[int, int, int], ...], font_sz_in_pts: float, + descriptor_for_idx: Callable[[int], Tuple[Union[FontObject | str], bool, bool]], + bold: int, + italic: int, + bold_italic: int, + num_symbol_fonts: int, + symbol_maps: Tuple[Tuple[int, int, int], ...], + font_sz_in_pts: float, narrow_symbols: Tuple[Tuple[int, int, int], ...], ) -> None: pass - def get_fallback_font(text: str, bold: bool, italic: bool) -> Any: pass - def create_test_font_group(sz: float, dpix: float, dpiy: float) -> tuple[int, int, int]: ... - class HistoryBuf: - def pagerhist_as_text(self, upto_output_start: bool = False) -> str: pass def pagerhist_as_bytes(self) -> bytes: pass - class LineBuf: - def is_continued(self, idx: int) -> bool: pass @@ -1213,7 +1028,6 @@ class LineBuf: def as_ansi(self, callback: Callable[[str], None]) -> None: ... - class Cursor: x: int y: int @@ -1225,9 +1039,7 @@ class Cursor: text_blink: bool shape: int - class Screen: - color_profile: ColorProfile columns: int lines: int @@ -1248,12 +1060,15 @@ class Screen: last_reported_cwd: Optional[bytes] def __init__( - self, - callbacks: Any = None, - lines: int = 80, columns: int = 24, scrollback: int = 0, - cell_width: int = 10, cell_height: int = 20, - window_id: int = 0, - test_child: Any = None + self, + callbacks: Any = None, + lines: int = 80, + columns: int = 24, + scrollback: int = 0, + cell_width: int = 10, + cell_height: int = 20, + window_id: int = 0, + test_child: Any = None, ): pass @@ -1264,14 +1079,13 @@ class Screen: def erase_last_command(self) -> bool: ... def set_progress(self, state: int, percent: int) -> None: ... def mark_potential_url_drag(self) -> bool: ... - def cursor_at_prompt(self) -> bool: pass def ignore_bells_for(self, duration: float = 1) -> None: pass - def set_window_char(self, ch: str = "") -> None: + def set_window_char(self, ch: str = '') -> None: pass def current_key_encoding_flags(self) -> int: @@ -1403,45 +1217,40 @@ class Screen: def delete_characters(self, num: int) -> None: ... def erase_characters(self, num: int) -> None: ... - def line_edge_colors(self) -> Tuple[int, int]: pass def current_pointer_shape(self) -> str: ... def change_pointer_shape(self, op: str, name: str) -> None: ... - def bell(self) -> None: ... def pause_rendering(self, pause: bool = True, for_how_long_in_ms: int = 100) -> bool: ... -def set_tab_bar_render_data( - os_window_id: int, screen: Screen, left: int, top: int, right: int, bottom: int -) -> None: +def set_tab_bar_render_data(os_window_id: int, screen: Screen, left: int, top: int, right: int, bottom: int) -> None: pass - -def set_window_title_bar_render_data( - os_window_id: int, tab_id: int, window_id: int, screen: Screen, - left: int, top: int, right: int, bottom: int -) -> None: +def set_window_title_bar_render_data(os_window_id: int, tab_id: int, window_id: int, screen: Screen, left: int, top: int, right: int, bottom: int) -> None: pass - def set_window_render_data( - os_window_id: int, tab_id: int, window_id: int, screen: Screen, - left: int, top: int, right: int, bottom: int, - spaces_left: int, spaces_top: int, spaces_right: int, spaces_bottom: int + os_window_id: int, + tab_id: int, + window_id: int, + screen: Screen, + left: int, + top: int, + right: int, + bottom: int, + spaces_left: int, + spaces_top: int, + spaces_right: int, + spaces_bottom: int, ) -> None: pass - -def truncate_point_for_length( - text: str, num_cells: int, start_pos: int = 0 -) -> int: +def truncate_point_for_length(text: str, num_cells: int, start_pos: int = 0) -> int: pass - class ChildMonitor: - def __init__( self, death_notify: Callable[[int, bool, int], None], @@ -1484,9 +1293,7 @@ class ChildMonitor: def inject_peer(self, fd: int) -> int: ... - class KeyEvent: - def __init__( self, key: int, shifted_key: int = 0, alternate_key: int = 0, mods: int = 0, action: int = 1, native_key: int = 1, ime_state: int = 0, text: str = '' ): @@ -1524,11 +1331,9 @@ class KeyEvent: def text(self) -> str: pass - def set_iutf8_fd(fd: int, on: bool) -> bool: pass - def spawn( exe: str, cwd: str, @@ -1547,51 +1352,35 @@ def spawn( ) -> int: pass - def set_window_drag_overlay(os_window_id: int, tab_id: int, window_id: int, quadrant: int) -> None: ... - - def set_window_padding(os_window_id: int, tab_id: int, window_id: int, left: int, top: int, right: int, bottom: int) -> None: pass - def click_mouse_url(os_window_id: int, tab_id: int, window_id: int) -> bool: pass - def click_mouse_cmd_output(os_window_id: int, tab_id: int, window_id: int, select_cmd_output: bool) -> bool: pass - def move_cursor_to_mouse_if_in_prompt(os_window_id: int, tab_id: int, window_id: int) -> bool: pass - def mouse_selection(os_window_id: int, tab_id: int, window_id: int, code: int, button: int) -> None: pass - def send_mouse_event( - screen: Screen, cell_x: int, cell_y: int, button: int, action: int, mods: int, - pixel_x: int = 0, pixel_y: int = 0, in_left_half_of_cell: bool = False + screen: Screen, cell_x: int, cell_y: int, button: int, action: int, mods: int, pixel_x: int = 0, pixel_y: int = 0, in_left_half_of_cell: bool = False ) -> bool: ... - - def set_window_logo(os_window_id: int, tab_id: int, window_id: int, path: str, position: str, alpha: float, png_data: bytes = b'') -> None: pass - -def get_window_logo_settings_if_not_default(os_window_id: int, tab_id: int, window_id: int) -> None | tuple[ - str, float, tuple[float, float, float, float]]: ... - +def get_window_logo_settings_if_not_default(os_window_id: int, tab_id: int, window_id: int) -> None | tuple[str, float, tuple[float, float, float, float]]: ... def apply_options_update() -> None: pass - def set_os_window_size(os_window_id: int, x: int, y: int) -> bool: pass - class OSWindowSize(TypedDict): width: int height: int @@ -1605,15 +1394,12 @@ class OSWindowSize(TypedDict): cell_height: int is_layer_shell: bool - def mark_os_window_dirty(os_window_id: int) -> None: pass - def get_os_window_size(os_window_id: int) -> Optional[OSWindowSize]: pass - def get_os_window_pos(os_window_id: int) -> Tuple[int, int]: pass @@ -1624,63 +1410,48 @@ def get_all_processes() -> Tuple[int, ...]: pass def glfw_get_monitor_names() -> tuple[tuple[str, str], ...]: ... - def num_users() -> int: pass - def redirect_mouse_handling(yes: bool) -> None: pass - def get_click_interval() -> float: pass - def glfw_get_keyboard_repeat_interval() -> float: pass - def send_data_to_peer(peer_id: int, data: Union[str, bytes], is_async_response: bool = False) -> None: pass - def set_os_window_title(os_window_id: int, title: str) -> None: pass - def get_os_window_title(os_window_id: int) -> Optional[str]: pass - def update_ime_position_for_window(window_id: int, force: bool = False, update_focus: int = 0) -> bool: pass - def shm_open(name: str, flags: int, mode: int = 0o600) -> int: pass - def shm_unlink(name: str) -> None: pass - def sigqueue(pid: int, signal: int, value: int) -> None: pass - def read_signals(fd: int, callback: Callable[[SignalInfo], None]) -> None: pass - def install_signal_handlers(*signals: int) -> Tuple[int, int]: pass - def remove_signal_handlers() -> None: pass - X25519: int SHA1_HASH: int SHA224_HASH: int @@ -1688,61 +1459,49 @@ SHA256_HASH: int SHA384_HASH: int SHA512_HASH: int - class Secret: pass - class EllipticCurveKey: - def __init__( - self, algorithm: int = 0 # X25519 - ): pass + self, + algorithm: int = 0, # X25519 + ): + pass def derive_secret( - self, pubkey: bytes, hash_algorithm: int = 0 # SHA256_HASH - ) -> Secret: pass + self, + pubkey: bytes, + hash_algorithm: int = 0, # SHA256_HASH + ) -> Secret: + pass @property def public(self) -> bytes: ... - @property def private(self) -> Secret: ... - class AES256GCMEncrypt: - def __init__(self, key: Secret): ... - def add_authenticated_but_unencrypted_data(self, data: bytes) -> None: ... - def add_data_to_be_encrypted(self, data: bytes, finished: bool = False) -> bytes: ... - @property def iv(self) -> bytes: ... - @property def tag(self) -> bytes: ... - class AES256GCMDecrypt: - def __init__(self, key: Secret, iv: bytes, tag: bytes): ... - def add_data_to_be_authenticated_but_not_decrypted(self, data: bytes) -> None: ... - def add_data_to_be_decrypted(self, data: bytes, finished: bool = False) -> bytes: ... - class Shlex: def __init__(self, src: str, allow_ansi_quoted_strings: bool = False): ... def next_word(self) -> Tuple[int, str]: ... def __next__(self) -> str: ... def __iter__(self) -> Iterator[str]: ... - class SingleKey: - __slots__ = () def __init__(self, mods: int = 0, is_native: object = False, key: int = -1): ... @@ -1761,7 +1520,6 @@ class SingleKey: def _replace(self, mods: int = 0, is_native: object = False, key: int = -1) -> 'SingleKey': ... def resolve_kitty_mod(self, mod: int) -> 'SingleKey': ... - def set_use_os_log(yes: bool) -> None: ... def get_docs_ref_map() -> bytes: ... def set_clipboard_data_types(ct: int, mime_types: Tuple[str, ...]) -> None: ... @@ -1790,11 +1548,11 @@ def is_layer_shell_supported() -> bool: ... def os_window_focus_counters() -> Dict[int, int]: ... def find_in_memoryview(buf: Union[bytes, memoryview, bytearray], chr: int) -> int: ... @overload -def replace_c0_codes_except_nl_space_tab(text: str) -> str:... +def replace_c0_codes_except_nl_space_tab(text: str) -> str: ... @overload -def replace_c0_codes_except_nl_space_tab(text: Union[bytes, memoryview, bytearray]) -> bytes:... -def terminfo_data() -> bytes:... -def wayland_compositor_data() -> Tuple[int, Optional[str]]:... +def replace_c0_codes_except_nl_space_tab(text: Union[bytes, memoryview, bytearray]) -> bytes: ... +def terminfo_data() -> bytes: ... +def wayland_compositor_data() -> Tuple[int, Optional[str]]: ... def monotonic() -> float: ... def timed_debug_print(x: str) -> None: ... def gpu_driver_version_string() -> str: ... @@ -1809,8 +1567,9 @@ def render_box_char(ch: int, width: int, height: int, scale: float = 1.0, dpi_x: def run_at_exit_cleanup_functions() -> None: ... def all_color_names() -> tuple[tuple[str, Color], ...]: ... def grab_keyboard(grab: bool | None) -> bool: ... -DecorationTypes = Literal[ - 'curl', 'dashed', 'dotted', 'double', 'straight', 'strikethrough', 'beam_cursor', 'underline_cursor', 'hollow_cursor', 'missing'] + +DecorationTypes = Literal['curl', 'dashed', 'dotted', 'double', 'straight', 'strikethrough', 'beam_cursor', 'underline_cursor', 'hollow_cursor', 'missing'] + def render_decoration( which: DecorationTypes, cell_width: int, cell_height: int, underline_position: int, underline_thickness: int, dpi: float = 96.0 ) -> bytes: ... @@ -1823,7 +1582,6 @@ class MousePosition(TypedDict): def get_mouse_data_for_window(os_window_id: int, tab_id: int, window_id: int) -> Optional[MousePosition]: ... - class StreamingBase64Decoder: # reset the state to empty to start decoding a new stream def reset(self) -> None: ... @@ -1834,7 +1592,6 @@ class StreamingBase64Decoder: # whether the data stream decoded so far is complete or not def needs_more_data(self) -> bool: ... - class StreamingBase64Encoder: def __init__(self, add_trailing_bytes: bool = True) -> None: ... # encode the specified data @@ -1844,10 +1601,8 @@ class StreamingBase64Encoder: # encode the specified data, return number of bytes written dest should be at least 4/3 *src + 2 bytes in size def encode_into(self, dest: WriteableBuffer, src: ReadableBuffer) -> int: ... - def start_drag_with_data( - os_window_id: int, data_map: dict[str, bytes], thumbnails: Sequence[tuple[bytes, int, int]], - operations: int = GLFW_DRAG_OPERATION_MOVE + os_window_id: int, data_map: dict[str, bytes], thumbnails: Sequence[tuple[bytes, int, int]], operations: int = GLFW_DRAG_OPERATION_MOVE ) -> None: ... def change_drag_thumbnail(os_window_id: int, idx: int = -1) -> None: ... def draw_single_line_of_text(os_window_id: int, text: str, fg: int, bg: int, width: int, padding_y: int = 2, max_width: bool = False) -> tuple[bytes, int]: ... @@ -1856,8 +1611,7 @@ def get_tab_being_dragged() -> tuple[int, bool, float, float]: ... def set_window_being_dragged(window_id: int = 0, drag_started: bool = False, x: float = 0.0, y: float = 0.0) -> None: ... def get_window_being_dragged() -> tuple[int, bool, float, float]: ... def request_callback_with_thumbnail( - callback: str, os_window_id: int, window_id: int = 0, include_tab_bar: bool = False, - scale: float = 0.25, max_width: int = 480 + callback: str, os_window_id: int, window_id: int = 0, include_tab_bar: bool = False, scale: float = 0.25, max_width: int = 480 ) -> None: ... def png_from_32bit_rgba_data(data: bytes, width: int, height: int, flip_vertically: bool = False) -> bytes: ... def set_uint_at_address(address: int, value: int) -> None: ... diff --git a/kitty/macos_process_info.c b/kitty/macos_process_info.c index 371009941..1612dde7f 100644 --- a/kitty/macos_process_info.c +++ b/kitty/macos_process_info.c @@ -286,12 +286,36 @@ error: } +static PyObject* +memory_of_process(PyObject *self UNUSED, PyObject *pid_) { + if (!PyLong_Check(pid_)) { PyErr_SetString(PyExc_TypeError, "pid must be an int"); return NULL; } + pid_t pid = (pid_t)PyLong_AsLong(pid_); + if (pid < 0) { PyErr_SetString(PyExc_TypeError, "pid cannot be negative"); return NULL; } + struct proc_taskinfo ti; + int ret = proc_pidinfo(pid, PROC_PIDTASKINFO, 0, &ti, sizeof(ti)); + if (ret <= 0) { PyErr_SetFromErrno(PyExc_OSError); return NULL; } + return PyLong_FromUnsignedLongLong(ti.pti_resident_size); +} + +static PyObject* +ppid_of_process(PyObject *self UNUSED, PyObject *pid_) { + if (!PyLong_Check(pid_)) { PyErr_SetString(PyExc_TypeError, "pid must be an int"); return NULL; } + pid_t pid = (pid_t)PyLong_AsLong(pid_); + if (pid < 0) { PyErr_SetString(PyExc_TypeError, "pid cannot be negative"); return NULL; } + struct proc_bsdshortinfo si; + int ret = proc_pidinfo(pid, PROC_PIDT_SHORTBSDINFO, 0, &si, sizeof(si)); + if (ret <= 0) { PyErr_SetFromErrno(PyExc_OSError); return NULL; } + return PyLong_FromUnsignedLong(si.pbsi_ppid); +} + static PyMethodDef module_methods[] = { {"cwd_of_process", (PyCFunction)cwd_of_process, METH_O, ""}, {"abspath_of_process", (PyCFunction)abspath_of_process, METH_O, ""}, {"cmdline_of_process", (PyCFunction)cmdline_of_process, METH_O, ""}, {"environ_of_process", (PyCFunction)environ_of_process, METH_O, ""}, {"get_all_processes", (PyCFunction)get_all_processes, METH_NOARGS, ""}, + {"memory_of_process", (PyCFunction)memory_of_process, METH_O, ""}, + {"ppid_of_process", (PyCFunction)ppid_of_process, METH_O, ""}, {NULL, NULL, 0, NULL} /* Sentinel */ }; diff --git a/kitty_tests/child.py b/kitty_tests/child.py new file mode 100644 index 000000000..921d6c0e7 --- /dev/null +++ b/kitty_tests/child.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python +# License: GPL v3 Copyright: 2026, Kovid Goyal + +import os +import subprocess + +from kitty.child import memory_used_by_process_tree_rooted_at +from kitty.constants import is_macos, kitty_exe + +from . import BaseTest + + +class ChildMemoryTest(BaseTest): + + def _spawn_allocating_child(self, alloc_bytes: int) -> subprocess.Popen: + p = subprocess.Popen( + [kitty_exe(), '+runpy', f'''\ +import sys, time +buf = bytearray({alloc_bytes}) +for i in range(0, {alloc_bytes}, 4096): + buf[i] = 1 +sys.stdout.write("ready\\n") +sys.stdout.flush() +time.sleep(300) +'''], + stdout=subprocess.PIPE, + ) + line = p.stdout.readline().strip() + p.stdout.close() + if line != b'ready': + p.kill() + p.wait() + raise AssertionError(f'Unexpected output from allocating child: {line!r}') + return p + + def _terminate(self, p: subprocess.Popen) -> None: + p.terminate() + p.wait() + + def test_memory_returns_positive_for_live_process(self): + mem = memory_used_by_process_tree_rooted_at(os.getpid()) + self.assertGreater(mem, 0) + + def test_memory_returns_minus_one_for_nonexistent_pid(self): + self.ae(memory_used_by_process_tree_rooted_at(99999999), -1) + + def test_memory_accounts_for_child_allocation(self): + # Verify that a child's known resident allocation shows up in the + # measurement. check_if_cgroup_root=True falls back to per-process + # tree walk when pid is not the cgroup root (the common case when + # running under a shared session cgroup), so this exercises the tree + # walk path on Linux and the always-tree-walk path on macOS. + alloc = 20 * 1024 * 1024 # 20 MiB + child = self._spawn_allocating_child(alloc) + try: + mem = memory_used_by_process_tree_rooted_at(child.pid, check_if_cgroup_root=True) + self.assertGreater( + mem, alloc // 2, + f'Expected at least {alloc // 2} bytes for a {alloc}-byte allocation, got {mem}', + ) + finally: + self._terminate(child) + + def test_memory_of_parent_tree_includes_child(self): + # Measuring the current process tree must yield more than measuring + # the child alone, because the test runner itself occupies memory. + alloc = 20 * 1024 * 1024 # 20 MiB + child = self._spawn_allocating_child(alloc) + try: + mem_child = memory_used_by_process_tree_rooted_at(child.pid, check_if_cgroup_root=True) + mem_tree = memory_used_by_process_tree_rooted_at(os.getpid(), check_if_cgroup_root=True) + self.assertGreater( + mem_tree, mem_child, + 'Parent tree memory should exceed child-only memory', + ) + finally: + self._terminate(child) + + def test_memory_cgroup_path_returns_positive(self): + # The fast cgroup path (check_if_cgroup_root=False, the default) must + # return a usable value on Linux. + if is_macos: + self.skipTest('cgroup not available on macOS') + mem = memory_used_by_process_tree_rooted_at(os.getpid()) + self.assertGreater(mem, 0) + + def test_memory_cgroup_and_tree_walk_both_positive(self): + # Both the cgroup path and the tree-walk path should give positive + # results for a process that is alive. + if is_macos: + self.skipTest('cgroup path not applicable on macOS') + alloc = 10 * 1024 * 1024 # 10 MiB + child = self._spawn_allocating_child(alloc) + try: + mem_cgroup = memory_used_by_process_tree_rooted_at(child.pid, check_if_cgroup_root=False) + mem_walk = memory_used_by_process_tree_rooted_at(child.pid, check_if_cgroup_root=True) + self.assertGreater(mem_cgroup, 0) + self.assertGreater(mem_walk, 0) + finally: + self._terminate(child)