refactor(langserver): remove async code from completions

Author: d-biehl

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

@@ -1,6 +1,6 @@
 import inspect
 from concurrent.futures import CancelledError, Future
-from threading import Thread, current_thread, local
+from threading import Event, Thread, current_thread, local
 from typing import Any, Callable, Dict, Generic, Optional, Tuple, TypeVar, cast, overload
 
 _F = TypeVar("_F", bound=Callable[..., Any])
@@ -12,10 +12,10 @@
 class FutureEx(Future, Generic[_TResult]):  # type: ignore[type-arg]
     def __init__(self) -> None:
         super().__init__()
-        self.cancelation_requested = False
+        self.cancelation_requested = Event()
 
     def cancel(self) -> bool:
-        self.cancelation_requested = True
+        self.cancelation_requested.set()
         return super().cancel()
 
     def result(self, timeout: Optional[float] = None) -> _TResult:
@@ -66,23 +66,36 @@ def _run_callable_in_thread_handler(
     _local_storage._local_future = future
     future.set_running_or_notify_cancel()
     try:
-        future.set_result(callable(*args, **kwargs))
+        result = callable(*args, **kwargs)
     except Exception as e:
         # TODO: add traceback to exception e.traceback = format_exc()
-
-        future.set_exception(e)
+        if not future.cancelled():
+            future.set_exception(e)
+    else:
+        if not future.cancelled():
+            future.set_result(result)
     finally:
         _local_storage._local_future = None
 
 
 def is_thread_cancelled() -> bool:
-    return _local_storage._local_future is not None and _local_storage._local_future.cancelation_requested
+    local_future = _local_storage._local_future
+    return local_future is not None and local_future.cancelation_requested.is_set()
+
+
+def check_thread_canceled(seconds: Optional[float] = None) -> None:
+    local_future = _local_storage._local_future
+    if local_future is None:
+        return
 
+    if seconds is None or seconds <= 0:
+        if not local_future.cancelation_requested.is_set():
+            return
+    elif not local_future.cancelation_requested.wait(seconds):
+        return
 
-def check_thread_canceled() -> None:
-    if _local_storage._local_future is not None and _local_storage._local_future.cancelation_requested:
-        name = current_thread().name
-        raise CancelledError(f"Thread {name+' ' if name else ' '}Cancelled")
+    name = current_thread().name
+    raise CancelledError(f"Thread {name+' ' if name else ' '}cancelled")
 
 
 def run_callable_in_thread(callable: Callable[..., _TResult], *args: Any, **kwargs: Any) -> FutureEx[_TResult]:

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

@@ -1,11 +1,8 @@
-from __future__ import annotations
-
-import asyncio
-from asyncio import CancelledError
+from concurrent.futures import CancelledError
 from itertools import chain
 from typing import TYPE_CHECKING, Any, Final, List, Optional, Union, cast
 
-from robotcode.core.async_tools import async_tasking_event
+from robotcode.core.event import event
 from robotcode.core.lsp.types import (
     CompletionContext,
     CompletionItem,
@@ -21,7 +18,7 @@
     TextEdit,
 )
 from robotcode.core.utils.logging import LoggingDescriptor
-from robotcode.core.utils.threading import threaded
+from robotcode.core.utils.threading import check_thread_canceled, threaded
 from robotcode.jsonrpc2.protocol import rpc_method
 from robotcode.language_server.common.decorators import (
     ALL_COMMIT_CHARACTERS_ATTR,
@@ -40,17 +37,17 @@
 class CompletionProtocolPart(LanguageServerProtocolPart):
     _logger: Final = LoggingDescriptor()
 
-    def __init__(self, parent: LanguageServerProtocol) -> None:
+    def __init__(self, parent: "LanguageServerProtocol") -> None:
         super().__init__(parent)
 
-    @async_tasking_event
-    async def collect(
+    @event
+    def collect(
         sender, document: TextDocument, position: Position, context: Optional[CompletionContext]  # NOSONAR
     ) -> Union[List[CompletionItem], CompletionList, None]:
         ...
 
-    @async_tasking_event
-    async def resolve(sender, completion_item: CompletionItem) -> Optional[CompletionItem]:  # NOSONAR
+    @event
+    def resolve(sender, completion_item: CompletionItem) -> Optional[CompletionItem]:  # NOSONAR
         ...
 
     def extend_capabilities(self, capabilities: ServerCapabilities) -> None:
@@ -84,7 +81,7 @@ def extend_capabilities(self, capabilities: ServerCapabilities) -> None:
 
     @rpc_method(name="textDocument/completion", param_type=CompletionParams)
     @threaded
-    async def _text_document_completion(
+    def _text_document_completion(
         self,
         text_document: TextDocumentIdentifier,
         position: Position,
@@ -95,21 +92,17 @@ async def _text_document_completion(
         results: List[Union[List[CompletionItem], CompletionList]] = []
 
         if context is not None and context.trigger_kind == CompletionTriggerKind.TRIGGER_CHARACTER:
-            await asyncio.sleep(0.25)
+            check_thread_canceled(0.25)
 
         document = self.parent.documents.get(text_document.uri)
         if document is None:
             return None
 
         p = document.position_from_utf16(position)
 
-        for result in await self.collect(
-            self,
-            document,
-            p,
-            context,
-            callback_filter=language_id_filter(document),
-        ):
+        for result in self.collect(self, document, p, context, callback_filter=language_id_filter(document)):
+            check_thread_canceled()
+
             if isinstance(result, BaseException):
                 if not isinstance(result, CancelledError):
                     self._logger.exception(result, exc_info=result)
@@ -121,6 +114,8 @@ async def _text_document_completion(
             return None
 
         for result in results:
+            check_thread_canceled()
+
             if isinstance(result, CompletionList):
                 for item in result.items:
                     if item.text_edit is not None:
@@ -155,15 +150,15 @@ def update_completion_item_to_utf16(self, document: TextDocument, item: Completi
 
     @rpc_method(name="completionItem/resolve", param_type=CompletionItem)
     @threaded
-    async def _completion_item_resolve(
+    def _completion_item_resolve(
         self,
         params: CompletionItem,
         *args: Any,
         **kwargs: Any,
     ) -> CompletionItem:
         results: List[CompletionItem] = []
 
-        for result in await self.resolve(self, params):
+        for result in self.resolve(self, params):
             if isinstance(result, BaseException):
                 if not isinstance(result, CancelledError):
                     self._logger.exception(result, exc_info=result)