fix(langserver): langserver that sometimes hangs, 'loop is not running' or 'lock take to long' messages

Rewrote and refactor diagnostics and refreshing part to use normal threads and not async. This should also bring a better performance for bigger projects, but maybe this needs some more optimizations.

Author: d-biehl

.vscode/launch.json

@@ -113,7 +113,9 @@
 				//"./tests/robotcode/language_server/robotframework/parts/test_discovering.py::test_workspace_discovery"
 				//"tests/robotcode/language_server/robotframework/parts/test_document_highlight.py"
 				//"tests/robotcode/utils/test_dataclasses.py::test_really_complex_data"
-				"tests/robotcode/language_server/robotframework/parts/test_foldingrange.py"
+				//"tests/robotcode/language_server/robotframework/parts/test_foldingrange.py"
+				//"tests/robotcode/language_server/robotframework/parts/test_document_symbols.py",
+				"./tests/robotcode/language_server/robotframework/parts/test_semantic_tokens.py "
 			],
 			"console": "integratedTerminal",
 			"justMyCode": false,

packages/core/src/robotcode/core/concurrent.py

@@ -1,20 +1,107 @@
+import contextlib
 import inspect
+import threading
 from concurrent.futures import CancelledError, Future
-from threading import Event, RLock, Thread, current_thread, local
+from types import TracebackType
 from typing import (
     Any,
     Callable,
     Dict,
     Generic,
+    Iterator,
     List,
     Optional,
+    Protocol,
     Tuple,
+    Type,
     TypeVar,
     cast,
     overload,
 )
 
-from typing_extensions import ParamSpec
+from typing_extensions import ParamSpec, Self
+
+
+class Lockable(Protocol):
+    def acquire(self, blocking: bool = True, timeout: float = -1) -> bool:
+        ...
+
+    def release(self) -> None:
+        ...
+
+    def __enter__(self) -> bool:
+        ...
+
+    def __exit__(
+        self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
+    ) -> None:
+        ...
+
+    def __str__(self) -> str:
+        ...
+
+
+class LockBase:
+    def __init__(
+        self,
+        lock: Lockable,
+        default_timeout: Optional[float] = None,
+        name: Optional[str] = None,
+    ) -> None:
+        self._default_timeout = default_timeout
+        self.name = name
+        self._lock = threading.RLock()
+
+    def acquire(self, blocking: bool = True, timeout: Optional[float] = None) -> bool:
+        timeout = timeout if timeout is not None else self._default_timeout or -1
+        aquired = self._lock.acquire(blocking, timeout=timeout)
+        if not aquired and blocking and timeout > 0:
+            raise RuntimeError(
+                f"Could not acquire {self.__class__.__qualname__} {self.name+' ' if self.name else ' '}in {timeout}s."
+            )
+        return aquired
+
+    def release(self) -> None:
+        return self._lock.release()
+
+    def __enter__(self) -> bool:
+        return self.acquire()
+
+    def __exit__(
+        self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
+    ) -> None:
+        return self.release()
+
+    def __str__(self) -> str:
+        return self._lock.__str__()
+
+    @contextlib.contextmanager
+    def __call__(self, *, timeout: Optional[float] = None) -> Iterator["Self"]:
+        aquired = self.acquire(timeout=timeout)
+        try:
+            yield self
+        finally:
+            if aquired:
+                self.release()
+
+
+class RLock(LockBase):
+    def __init__(
+        self,
+        default_timeout: Optional[float] = None,
+        name: Optional[str] = None,
+    ) -> None:
+        super().__init__(threading.RLock(), default_timeout=default_timeout, name=name)
+
+
+class Lock(LockBase):
+    def __init__(
+        self,
+        default_timeout: Optional[float] = None,
+        name: Optional[str] = None,
+    ) -> None:
+        super().__init__(threading.Lock(), default_timeout=default_timeout, name=name)
+
 
 _F = TypeVar("_F", bound=Callable[..., Any])
 _TResult = TypeVar("_TResult")
@@ -25,7 +112,7 @@
 class Task(Future, Generic[_TResult]):  # type: ignore[type-arg]
     def __init__(self) -> None:
         super().__init__()
-        self.cancelation_requested_event = Event()
+        self.cancelation_requested_event = threading.Event()
 
     @property
     def cancelation_requested(self) -> bool:
@@ -43,16 +130,16 @@ def add_done_callback(self, fn: Callable[["Task[Any]"], Any]) -> None:
 
 
 @overload
-def threaded(__func: _F) -> _F:
+def threaded_task(__func: _F) -> _F:
     ...
 
 
 @overload
-def threaded(*, enabled: bool = True) -> Callable[[_F], _F]:
+def threaded_task(*, enabled: bool = True) -> Callable[[_F], _F]:
     ...
 
 
-def threaded(__func: _F = None, *, enabled: bool = True) -> Callable[[_F], _F]:
+def threaded_task(__func: _F = None, *, enabled: bool = True) -> Callable[[_F], _F]:
     def decorator(func: _F) -> _F:
         setattr(func, __THREADED_MARKER, enabled)
         return func
@@ -71,7 +158,7 @@ def is_threaded_callable(callable: Callable[..., Any]) -> bool:
     )
 
 
-class _Local(local):
+class _Local(threading.local):
     def __init__(self) -> None:
         super().__init__()
         self._local_future: Optional[Task[Any]] = None
@@ -119,14 +206,14 @@ def check_current_task_canceled(at_least_seconds: Optional[float] = None, raise_
         return False
 
     if raise_exception:
-        name = current_thread().name
+        name = threading.current_thread().name
         raise CancelledError(f"Thread {name + ' ' if name else ' '}cancelled")
 
     return True
 
 
 _running_tasks_lock = RLock()
-_running_tasks: Dict[Task[Any], Thread] = {}
+_running_tasks: Dict[Task[Any], threading.Thread] = {}
 
 
 def _remove_future_from_running_tasks(future: Task[Any]) -> None:
@@ -140,7 +227,7 @@ def _remove_future_from_running_tasks(future: Task[Any]) -> None:
 def run_as_task(callable: Callable[_P, _TResult], *args: _P.args, **kwargs: _P.kwargs) -> Task[_TResult]:
     future: Task[_TResult] = Task()
     with _running_tasks_lock:
-        thread = Thread(
+        thread = threading.Thread(
             target=_run_task_in_thread_handler,
             args=(future, callable, args, kwargs),
             name=str(callable),
@@ -155,12 +242,12 @@ def run_as_task(callable: Callable[_P, _TResult], *args: _P.args, **kwargs: _P.k
 
 
 def _cancel_all_running_tasks(timeout: Optional[float] = None) -> None:
-    threads: List[Thread] = []
+    threads: List[threading.Thread] = []
     with _running_tasks_lock:
         for future, thread in _running_tasks.items():
             if not future.cancelation_requested:
                 future.cancel()
                 threads.append(thread)
     for thread in threads:
-        if thread is not current_thread():
+        if thread is not threading.current_thread():
             thread.join(timeout=timeout)

packages/core/src/robotcode/core/event.py

@@ -31,16 +31,16 @@ def __init__(self) -> None:
 
         self._listeners: MutableSet[weakref.ref[Any]] = set()
 
-    def add(self, callback: Callable[_TParams, _TResult]) -> None:
-        def remove_listener(ref: Any) -> None:
-            with self._lock:
-                self._listeners.remove(ref)
+    def __remove_listener(self, ref: Any) -> None:
+        with self._lock:
+            self._listeners.remove(ref)
 
+    def add(self, callback: Callable[_TParams, _TResult]) -> None:
         with self._lock:
             if inspect.ismethod(callback):
-                self._listeners.add(weakref.WeakMethod(callback, remove_listener))
+                self._listeners.add(weakref.WeakMethod(callback, self.__remove_listener))
             else:
-                self._listeners.add(weakref.ref(callback, remove_listener))
+                self._listeners.add(weakref.ref(callback, self.__remove_listener))
 
     def remove(self, callback: Callable[_TParams, _TResult]) -> None:
         with self._lock:
@@ -155,7 +155,3 @@ def __init__(self, _func: Callable[_TParams, _TResult]) -> None:
 class event(EventDescriptorBase[_TParams, _TResult, Event[_TParams, _TResult]]):  # noqa: N801
     def __init__(self, _func: Callable[_TParams, _TResult]) -> None:
         super().__init__(_func, Event[_TParams, _TResult])
-
-
-# TODO implement this as something with concurrent.futures.Future
-tasking_event = event

packages/core/src/robotcode/core/utils/logging.py

@@ -498,8 +498,8 @@ def build_exception_message(exception: BaseException) -> str:
 
                 result = None
                 try:
-                    start_time: float = 0
-                    end_time: float = 0
+                    start_time: float = 0.0
+                    end_time: float = 0.0
                     if timed:
                         start_time = time.monotonic()
 

packages/jsonrpc2/src/robotcode/jsonrpc2/protocol.py

@@ -868,15 +868,12 @@ async def handle_notification(self, message: JsonRPCNotification) -> None:
 
 class GenericJsonRPCProtocolPart(Generic[TProtocol]):
     def __init__(self, parent: TProtocol) -> None:
-        self._parent = weakref.ref(parent)
+        self._parent = parent
         parent.registry.add_class_part_instance(self)
 
     @property
     def parent(self) -> TProtocol:
-        result = self._parent()
-        if result is None:
-            raise JsonRPCException("WeakRef is dead.")
-        return result
+        return self._parent
 
 
 class JsonRPCProtocolPart(GenericJsonRPCProtocolPart[JsonRPCProtocol]):

packages/language_server/src/robotcode/language_server/common/parts/code_lens.py

@@ -1,7 +1,8 @@
+import threading
 from concurrent.futures import CancelledError
 from typing import TYPE_CHECKING, Any, Final, List, Optional
 
-from robotcode.core.concurrent import Task, check_current_task_canceled, run_as_task
+from robotcode.core.concurrent import check_current_task_canceled
 from robotcode.core.event import event
 from robotcode.core.lsp.types import (
     CodeLens,
@@ -26,8 +27,8 @@ class CodeLensProtocolPart(LanguageServerProtocolPart):
 
     def __init__(self, parent: "LanguageServerProtocol") -> None:
         super().__init__(parent)
-        self.refresh_task: Optional[Task[Any]] = None
-        self._refresh_timeout = 5
+        self.refresh_timer_lock = threading.RLock()
+        self.refresh_timer: Optional[threading.Timer] = None
 
     @event
     def collect(sender, document: TextDocument) -> Optional[List[CodeLens]]:
@@ -88,20 +89,27 @@ def _code_lens_resolve(self, params: CodeLens, *args: Any, **kwargs: Any) -> Cod
 
         return params
 
-    def refresh(self, now: bool = True) -> None:
-        if self.refresh_task is not None and not self.refresh_task.done():
-            self.refresh_task.cancel()
+    def refresh(self, now: bool = False) -> None:
+        with self.refresh_timer_lock:
+            if self.refresh_timer is not None:
+                self.refresh_timer.cancel()
+                self.refresh_timer = None
 
-        self.refresh_task = run_as_task(self._refresh, now)
+            if not now:
+                self.refresh_timer = threading.Timer(1, self._refresh)
+                self.refresh_timer.start()
+                return
+
+        self._refresh()
+
+    def _refresh(self) -> None:
+        with self.refresh_timer_lock:
+            self.refresh_timer = None
 
-    def _refresh(self, now: bool = True) -> None:
         if (
             self.parent.client_capabilities is not None
             and self.parent.client_capabilities.workspace is not None
             and self.parent.client_capabilities.workspace.code_lens is not None
             and self.parent.client_capabilities.workspace.code_lens.refresh_support
         ):
-            if not now:
-                check_current_task_canceled(1)
-
-            self.parent.send_request("workspace/codeLens/refresh").result(self._refresh_timeout)
+            self.parent.send_request("workspace/codeLens/refresh")