limit concurrent diagnostics

Author: seeM

jedi_language_server/constants.py

@@ -1 +1,4 @@
 """Constants."""
+
+MAX_CONCURRENT_DEBOUNCE_CALLS = 10
+"""The maximum number of concurrent calls allowed by the debounce decorator."""

jedi_language_server/jedi_utils.py

@@ -41,6 +41,7 @@
 )
 from pygls.workspace import TextDocument
 
+from .constants import MAX_CONCURRENT_DEBOUNCE_CALLS
 from .initialization_options import HoverDisableOptions, InitializationOptions
 from .type_map import get_lsp_completion_type, get_lsp_symbol_type
 
@@ -53,13 +54,15 @@
 P = ParamSpec("P")
 
 
+_debounce_semaphore = threading.Semaphore(MAX_CONCURRENT_DEBOUNCE_CALLS)
+
+
 def debounce(
-    interval_s: int, keyed_by: Optional[str] = None
+    interval_s: float, keyed_by: Optional[str] = None
 ) -> Callable[[Callable[P, None]], Callable[P, None]]:
     """Debounce calls to this function until interval_s seconds have passed.
 
-    Decorator copied from https://github.com/python-lsp/python-lsp-
-    server
+    Decorator adapted from https://github.com/python-lsp/python-lsp-server
     """
 
     def wrapper(func: Callable[P, None]) -> Callable[P, None]:
@@ -68,20 +71,23 @@ def wrapper(func: Callable[P, None]) -> Callable[P, None]:
 
         @functools.wraps(func)
         def debounced(*args: P.args, **kwargs: P.kwargs) -> None:
+            _debounce_semaphore.acquire()
+
             sig = inspect.signature(func)
             call_args = sig.bind(*args, **kwargs)
             key = call_args.arguments[keyed_by] if keyed_by else None
 
             def run() -> None:
                 with lock:
                     del timers[key]
-                return func(*args, **kwargs)
+                func(*args, **kwargs)
+                _debounce_semaphore.release()
 
             with lock:
                 old_timer = timers.get(key)
                 if old_timer:
                     old_timer.cancel()
-
+                    _debounce_semaphore.release()
                 timer = threading.Timer(interval_s, run)
                 timers[key] = timer
                 timer.start()

tests/test_debounce.py

@@ -0,0 +1,39 @@
+import threading
+import time
+from collections import Counter
+
+from jedi_language_server.constants import MAX_CONCURRENT_DEBOUNCE_CALLS
+from jedi_language_server.jedi_utils import debounce
+
+
+def test_debounce() -> None:
+    """Test that the debounce decorator delays and limits the number of concurrent calls."""
+    # Create a function that records call counts per URI.
+    counter = Counter[str]()
+    cond = threading.Condition()
+
+    def f(uri: str) -> None:
+        with cond:
+            counter.update(uri)
+            cond.notify_all()
+
+    # Call the debounced function more than the max allowed concurrent calls.
+    debounced = debounce(interval_s=1, keyed_by="uri")(f)
+    for _ in range(3):
+        debounced("0")
+    for i in range(1, MAX_CONCURRENT_DEBOUNCE_CALLS + 10):
+        debounced(str(i))
+
+    # Wait for at least the max allowed concurrent timers to complete.
+    with cond:
+        assert cond.wait_for(
+            lambda: sum(counter.values()) >= MAX_CONCURRENT_DEBOUNCE_CALLS
+        )
+
+    # Check the counter after 0.5 seconds to ensure that no additional timers
+    # were started and have completed.
+    time.sleep(0.5)
+    assert sum(counter.values()) == MAX_CONCURRENT_DEBOUNCE_CALLS
+
+    # For uri "0", only one timer should have been started despite 3 calls.
+    assert counter["0"] == 1