diff --git a/.github/workflows/unittests.yml b/.github/workflows/unittests.yml index ca3a6af..0724570 100644 --- a/.github/workflows/unittests.yml +++ b/.github/workflows/unittests.yml @@ -29,4 +29,4 @@ jobs: - name: Test run: - FAKE_ENV=1 pytest-3 -v --disable-pytest-warnings tests/ + PYTHONPATH=. FAKE_ENV=1 pytest-3 -v --disable-pytest-warnings tests/ diff --git a/midcli/__main__.py b/midcli/__main__.py index 22bc16b..2196ef4 100644 --- a/midcli/__main__.py +++ b/midcli/__main__.py @@ -16,6 +16,7 @@ from prompt_toolkit.layout.processors import ( ConditionalProcessor, HighlightMatchingBracketProcessor, TabsProcessor, ) +from prompt_toolkit.input.typeahead import clear_typeahead from prompt_toolkit.shortcuts import PromptSession, CompleteStyle from prompt_toolkit.styles import Style @@ -30,6 +31,7 @@ from .menu.items import get_menu_items, process_menu_item from .pager import enable_pager from .utils.shell import is_main_cli, switch_to_shell +from .utils.throttle import Throttler class CLI: @@ -235,7 +237,15 @@ def run(self): menu_app = self._build_menu(menu_items) prompt_app = None + # Drop input that piled up in prompt_toolkit's typeahead buffer + # while we were sleeping (e.g. a key held down), so we don't replay + # the backlog the moment it's released. Both sessions share stdin, + # so clearing via either input clears the buffer for both. + throttler = Throttler(flush=lambda: clear_typeahead(menu_app.input)) + while True: + throttler.throttle() + if self.context.menu_item: process_menu_item(self.context, menu_items, self.context.menu_item) break diff --git a/midcli/utils/throttle.py b/midcli/utils/throttle.py new file mode 100644 index 0000000..ef5c9c0 --- /dev/null +++ b/midcli/utils/throttle.py @@ -0,0 +1,45 @@ +# -*- coding=utf-8 -*- +import time + +__all__ = ["Throttler"] + + +class Throttler: + """ + Prevents a tight loop from spinning the CPU when its body keeps returning + immediately (e.g. a prompt that gets EOF/empty input repeatedly). + + The first `burst` calls to `throttle()` return immediately. After that, each + call sleeps for `delay` seconds. If more than `reset_after` seconds elapse + between two consecutive calls, the throttle is reset and the next `burst` + calls are instantaneous again. + + `flush` is called once after every sleep. It is meant to discard whatever + input piled up while we were sleeping (e.g. prompt_toolkit's typeahead + buffer when a key is held down), so the caller doesn't replay a backlog the + moment the key is released. + """ + + def __init__(self, *, burst=10, delay=0.1, reset_after=1.0, flush=None, + monotonic=time.monotonic, sleep=time.sleep): + self.burst = burst + self.delay = delay + self.reset_after = reset_after + self._flush = flush + self._monotonic = monotonic + self._sleep = sleep + self._count = 0 + self._last = None + + def throttle(self): + now = self._monotonic() + if self._last is not None and now - self._last > self.reset_after: + self._count = 0 + self._last = now + + if self._count >= self.burst: + self._sleep(self.delay) + if self._flush is not None: + self._flush() + else: + self._count += 1 diff --git a/tests/utils/test_throttle.py b/tests/utils/test_throttle.py new file mode 100644 index 0000000..abacf73 --- /dev/null +++ b/tests/utils/test_throttle.py @@ -0,0 +1,111 @@ +# -*- coding=utf-8 -*- +from unittest.mock import Mock + +from midcli.utils.throttle import Throttler + + +def make_throttler(**kwargs): + """Throttler driven by a virtual clock so tests don't actually sleep.""" + clock = {"now": 0.0} + sleeps = [] + flushes = [] + + def monotonic(): + return clock["now"] + + def sleep(seconds): + sleeps.append(seconds) + + kwargs.setdefault("flush", lambda: flushes.append(clock["now"])) + throttler = Throttler(monotonic=monotonic, sleep=sleep, **kwargs) + throttler._test_flushes = flushes + return throttler, clock, sleeps + + +def test_first_burst_calls_do_not_sleep(): + throttler, clock, sleeps = make_throttler(burst=10, delay=0.01) + + for _ in range(10): + throttler.throttle() + + assert sleeps == [] + # No sleep -> nothing accumulated -> no flush. + assert throttler._test_flushes == [] + + +def test_calls_after_burst_sleep(): + throttler, clock, sleeps = make_throttler(burst=10, delay=0.01) + + for _ in range(15): + throttler.throttle() + + # First 10 are instantaneous, the remaining 5 each sleep `delay`. + assert sleeps == [0.01] * 5 + # Each sleep is followed by a flush of the accumulated input. + assert len(throttler._test_flushes) == 5 + + +def test_gap_longer_than_reset_resets_throttle(): + throttler, clock, sleeps = make_throttler(burst=10, delay=0.01, reset_after=1.0) + + # Exhaust the burst so the next call would sleep. + for _ in range(15): + throttler.throttle() + assert sleeps == [0.01] * 5 + + # More than a second passes between two calls -> throttle resets. + clock["now"] += 1.5 + sleeps.clear() + + for _ in range(10): + throttler.throttle() + + assert sleeps == [] + + +def test_gap_within_reset_does_not_reset_throttle(): + throttler, clock, sleeps = make_throttler(burst=10, delay=0.01, reset_after=1.0) + + for _ in range(10): + throttler.throttle() + assert sleeps == [] + + # Exactly one second is not "more than a second", so no reset. + clock["now"] += 1.0 + throttler.throttle() + + assert sleeps == [0.01] + + +def test_flush_runs_after_each_sleep_only(): + flushes = [] + throttler, clock, sleeps = make_throttler( + burst=2, delay=0.01, flush=lambda: flushes.append(True), + ) + + # First 2 calls: no sleep, no flush. + throttler.throttle() + throttler.throttle() + assert flushes == [] + + # 3rd call sleeps and then flushes. + throttler.throttle() + assert flushes == [True] + + +def test_no_flush_callback_is_fine(): + throttler = Throttler(burst=1, monotonic=lambda: 0.0, sleep=lambda s: None) + + throttler.throttle() + throttler.throttle() # would sleep+flush; flush is None, must not raise + + +def test_uses_real_sleep_and_monotonic_by_default(): + sleep = Mock() + monotonic = Mock(return_value=0.0) + throttler = Throttler(burst=1, monotonic=monotonic, sleep=sleep) + + throttler.throttle() + throttler.throttle() + + sleep.assert_called_once_with(throttler.delay)