Improve performance of using active process data when rendering the tab bar by only scanning processes once per second

We dont bother with configurable ttl. Instead treat the start of caching
as the instant when cache freshness is checked. And ensure that cache is
re-used for every OS Window.

Fixes #9862
Fixes #9872
This commit is contained in:
Kovid Goyal 2026-04-15 07:53:09 +05:30
parent 15a6a51a22
commit f65438d6a5
No known key found for this signature in database
GPG key ID: 06BC317B515ACE7C
4 changed files with 48 additions and 15 deletions

View file

@ -243,6 +243,8 @@ Detailed list of changes
- Render block elements from the Unicode Symbols for Legacy Computing Supplement block (U+1CC00U+1CEBF): separated block quadrants, separated block sextants, one sixteenth blocks, and one quarter block partial fills (:disc:`9849`)
- Improve performance of using active process data when rendering the tab bar by only scanning processes once per second (:iss:`9862`)
0.46.2 [2026-03-21]
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

View file

@ -27,7 +27,7 @@
from kitty.types import WindowResizeDrag
from .child import cached_process_data, default_env, set_default_env
from .child import cached_process_data, default_env, process_data_cache, set_default_env
from .cli import create_opts, green, parse_args
from .cli_stub import CLIOptions, SaveAsSessionOptions
from .clipboard import (
@ -1923,6 +1923,13 @@ def on_activity_since_last_focus(self, window: Window) -> None:
if tm is not None:
tm.mark_tab_bar_dirty()
def cache_process_data(self, enable: bool) -> None:
' Turn on caching of process data. Must be called in enable/disable pairs. '
if enable:
self.process_data_cache_active = process_data_cache.start_caching()
else:
process_data_cache.stop_caching(self.process_data_cache_active)
def update_tab_bar_data(self, os_window_id: int) -> None:
tm = self.os_window_map.get(os_window_id)
if tm is not None:

View file

@ -1001,6 +1001,7 @@ render(monotonic_t now, bool input_read) {
const bool scan_for_animated_images = global_state.check_for_active_animated_images;
global_state.check_for_active_animated_images = false;
call_boss(cache_process_data, "O", Py_True);
for (size_t i = 0; i < global_state.num_os_windows; i++) {
OSWindow *w = global_state.os_windows + i;
@ -1021,6 +1022,7 @@ render(monotonic_t now, bool input_read) {
}
last_render_at = now;
call_boss(cache_process_data, "O", Py_False);
#undef TD
}

View file

@ -8,6 +8,7 @@
from collections.abc import Generator, Sequence
from contextlib import contextmanager, suppress
from itertools import count
from time import monotonic
from typing import TYPE_CHECKING, DefaultDict, Iterable, Mapping, Optional, TypedDict
import kitty.fast_data_types as fast_data_types
@ -96,27 +97,48 @@ def checked_terminfo_dir() -> str | None:
return terminfo_dir if os.path.isdir(terminfo_dir) else None
def processes_in_group(grp: int) -> list[int]:
gmap: DefaultDict[int, list[int]] | None = getattr(process_group_map, 'cached_map', None)
if gmap is None:
try:
gmap = process_group_map()
except Exception:
gmap = defaultdict(list)
return gmap.get(grp, [])
class CachedProcessData:
cached_result: DefaultDict[int, list[int]] | None = None
cache_active: bool = False
cache_at: float = 0
ttl: float = 1
def process_group_map(self) -> DefaultDict[int, list[int]]:
if self.cached_result is None or not self.cache_active:
try:
self.cached_result = process_group_map()
except Exception:
self.cached_result = defaultdict(list)
self.cache_at = monotonic()
return self.cached_result
def processes_in_group(self, grp: int) -> list[int]:
return self.process_group_map()[grp]
def start_caching(self, refresh: bool = False) -> bool:
prev, self.cache_active = self.cache_active, True
if refresh or monotonic() - self.cache_at > self.ttl:
self.cached_result = None
self.cache_at = 0
return prev
def stop_caching(self, prev: bool) -> None:
self.cache_active = prev
self.cached_result = None
process_data_cache = CachedProcessData()
processes_in_group = process_data_cache.processes_in_group
@contextmanager
def cached_process_data() -> Generator[None, None, None]:
try:
cm = process_group_map()
except Exception:
cm = defaultdict(list)
setattr(process_group_map, 'cached_map', cm)
orig = process_data_cache.start_caching(refresh=True)
try:
yield
finally:
delattr(process_group_map, 'cached_map')
process_data_cache.stop_caching(orig)
def session_id(pids: Iterable[int]) -> int: