NAS-141550 / 27.0.0-BETA.1 / Throttle input if a key is stuck

.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/

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

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

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)