refactor and reimplement running coroutines in separate threads

Author: d-biehl

robotcode/debugger/launcher/server.py

@@ -73,7 +73,7 @@ def connected(self) -> bool:
         return self._client is not None and self._client.connected
 
     @rpc_method(name="initialize", param_type=InitializeRequestArguments)
-    async def _initialize(self, arguments: InitializeRequestArguments) -> Capabilities:
+    async def _initialize(self, arguments: InitializeRequestArguments, *args: Any, **kwargs: Any) -> Capabilities:
         self._initialized = True
 
         return Capabilities(
@@ -141,7 +141,8 @@ async def _launch(
         groupOutput: Optional[bool] = False,
         stopOnEntry: Optional[bool] = False,  # noqa: N803
         arguments: Optional[LaunchRequestArguments] = None,
-        **kwargs: Any,
+        *_args: Any,
+        **_kwargs: Any,
     ) -> None:
         from ...utils.net import find_free_port
 
@@ -235,11 +236,13 @@ async def _launch(
             raise asyncio.TimeoutError("Can't connect to debug launcher.")
 
     @rpc_method(name="configurationDone", param_type=ConfigurationDoneArguments)
-    async def _configuration_done(self, arguments: Optional[ConfigurationDoneArguments] = None) -> None:
+    async def _configuration_done(
+        self, arguments: Optional[ConfigurationDoneArguments] = None, *args: Any, **kwargs: Any
+    ) -> None:
         await self.client.protocol.send_request_async(ConfigurationDoneRequest(arguments=arguments))
 
     @rpc_method(name="disconnect", param_type=DisconnectArguments)
-    async def _disconnect(self, arguments: Optional[DisconnectArguments] = None) -> None:
+    async def _disconnect(self, arguments: Optional[DisconnectArguments] = None, *args: Any, **kwargs: Any) -> None:
         if self.connected:
             if not self.client.protocol.terminated:
                 await self.client.protocol.send_request_async(DisconnectRequest(arguments=arguments))
@@ -248,7 +251,7 @@ async def _disconnect(self, arguments: Optional[DisconnectArguments] = None) ->
 
     @_logger.call
     @rpc_method(name="terminate", param_type=TerminateArguments)
-    async def _terminate(self, arguments: Optional[TerminateArguments] = None) -> None:
+    async def _terminate(self, arguments: Optional[TerminateArguments] = None, *args: Any, **kwargs: Any) -> None:
         if self.client.connected:
             return await self.client.protocol.send_request_async(TerminateRequest(arguments=arguments))
         else:

robotcode/debugger/server.py

@@ -124,7 +124,7 @@ async def terminate(self) -> None:
             self._terminated = True
 
     @rpc_method(name="terminate", param_type=TerminateArguments)
-    async def _terminate(self, arguments: Optional[TerminateArguments] = None) -> None:
+    async def _terminate(self, arguments: Optional[TerminateArguments] = None, *args: Any, **kwargs: Any) -> None:
         import signal
 
         if not self._sigint_signaled:
@@ -138,13 +138,15 @@ async def _terminate(self, arguments: Optional[TerminateArguments] = None) -> No
         Debugger.instance().stop()
 
     @rpc_method(name="disconnect", param_type=DisconnectArguments)
-    async def _disconnect(self, arguments: Optional[DisconnectArguments] = None) -> None:
+    async def _disconnect(self, arguments: Optional[DisconnectArguments] = None, *args: Any, **kwargs: Any) -> None:
         if not (await self.exited) or not (await self.terminated):
             if arguments is None or arguments.terminate_debuggee is None or arguments.terminate_debuggee:
                 os._exit(-1)
 
     @rpc_method(name="setBreakpoints", param_type=SetBreakpointsArguments)
-    async def _set_breakpoints(self, arguments: SetBreakpointsArguments) -> SetBreakpointsResponseBody:
+    async def _set_breakpoints(
+        self, arguments: SetBreakpointsArguments, *args: Any, **kwargs: Any
+    ) -> SetBreakpointsResponseBody:
         return SetBreakpointsResponseBody(
             breakpoints=Debugger.instance().set_breakpoints(
                 arguments.source, arguments.breakpoints, arguments.lines, arguments.source_modified
@@ -153,7 +155,9 @@ async def _set_breakpoints(self, arguments: SetBreakpointsArguments) -> SetBreak
 
     @_logger.call
     @rpc_method(name="configurationDone", param_type=ConfigurationDoneArguments)
-    async def _configuration_done(self, arguments: Optional[ConfigurationDoneArguments] = None) -> None:
+    async def _configuration_done(
+        self, arguments: Optional[ConfigurationDoneArguments] = None, *args: Any, **kwargs: Any
+    ) -> None:
         self._received_configuration_done = True
         self._received_configuration_done_event.set()
 
@@ -164,39 +168,39 @@ async def wait_for_configuration_done(self, timeout: float = 5) -> bool:
         return self._received_configuration_done
 
     @rpc_method(name="continue", param_type=ContinueArguments)
-    async def _continue(self, arguments: ContinueArguments) -> ContinueResponseBody:
+    async def _continue(self, arguments: ContinueArguments, *args: Any, **kwargs: Any) -> ContinueResponseBody:
         Debugger.instance().continue_thread(arguments.thread_id)
         return ContinueResponseBody(all_threads_continued=True)
 
     @rpc_method(name="pause", param_type=PauseArguments)
-    async def _pause(self, arguments: PauseArguments) -> None:
+    async def _pause(self, arguments: PauseArguments, *args: Any, **kwargs: Any) -> None:
         Debugger.instance().pause_thread(arguments.thread_id)
 
     @rpc_method(name="next", param_type=NextArguments)
-    async def _next(self, arguments: NextArguments) -> None:
+    async def _next(self, arguments: NextArguments, *args: Any, **kwargs: Any) -> None:
         Debugger.instance().next(arguments.thread_id, arguments.granularity)
 
     @rpc_method(name="stepIn", param_type=StepInArguments)
-    async def _step_in(self, arguments: StepInArguments) -> None:
+    async def _step_in(self, arguments: StepInArguments, *args: Any, **kwargs: Any) -> None:
         Debugger.instance().step_in(arguments.thread_id, arguments.target_id, arguments.granularity)
 
     @rpc_method(name="stepOut", param_type=StepOutArguments)
-    async def _step_out(self, arguments: StepOutArguments) -> None:
+    async def _step_out(self, arguments: StepOutArguments, *args: Any, **kwargs: Any) -> None:
         Debugger.instance().step_out(arguments.thread_id, arguments.granularity)
 
     @rpc_method(name="threads")
-    async def _threads(self) -> ThreadsResponseBody:
+    async def _threads(self, *args: Any, **kwargs: Any) -> ThreadsResponseBody:
         return ThreadsResponseBody(threads=Debugger.instance().get_threads())
 
     @rpc_method(name="stackTrace", param_type=StackTraceArguments)
-    async def _stack_trace(self, arguments: StackTraceArguments) -> StackTraceResponseBody:
+    async def _stack_trace(self, arguments: StackTraceArguments, *args: Any, **kwargs: Any) -> StackTraceResponseBody:
         result = Debugger.instance().get_stack_trace(
             arguments.thread_id, arguments.start_frame, arguments.levels, arguments.format
         )
         return StackTraceResponseBody(stack_frames=result.stack_frames, total_frames=result.total_frames)
 
     @rpc_method(name="scopes", param_type=ScopesArguments)
-    async def _scopes(self, arguments: ScopesArguments) -> ScopesResponseBody:
+    async def _scopes(self, arguments: ScopesArguments, *args: Any, **kwargs: Any) -> ScopesResponseBody:
         return ScopesResponseBody(scopes=Debugger.instance().get_scopes(arguments.frame_id))
 
     @rpc_method(name="variables", param_type=VariablesArguments)
@@ -208,6 +212,8 @@ async def _variables(
         start: Optional[int] = None,
         count: Optional[int] = None,
         format: Optional[ValueFormat] = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> VariablesResponseBody:
         return VariablesResponseBody(
             variables=Debugger.instance().get_variables(variables_reference, filter, start, count, format)
@@ -221,6 +227,8 @@ async def _evaluate(
         frame_id: Optional[int] = None,
         context: Union[EvaluateArgumentContext, str, None] = None,
         format: Optional[ValueFormat] = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> EvaluateResponseBody:
         result = Debugger.instance().evaluate(expression, frame_id, context, format)
         return EvaluateResponseBody(
@@ -241,6 +249,8 @@ async def _set_variable(
         name: str,
         value: str,
         format: Optional[ValueFormat] = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> SetVariableResponseBody:
         result = Debugger.instance().set_variable(variables_reference, name, value, format)
         return SetVariableResponseBody(
@@ -253,7 +263,7 @@ async def _set_variable(
 
     @rpc_method(name="setExceptionBreakpoints", param_type=SetExceptionBreakpointsArguments)
     async def _set_exception_breakpoints(
-        self, arguments: SetExceptionBreakpointsArguments
+        self, arguments: SetExceptionBreakpointsArguments, *args: Any, **kwargs: Any
     ) -> Optional[SetExceptionBreakpointsResponseBody]:
         result = Debugger.instance().set_exception_breakpoints(
             arguments.filters, arguments.filter_options, arguments.exception_options

robotcode/jsonrpc2/protocol.py

@@ -31,7 +31,7 @@
     runtime_checkable,
 )
 
-from ..utils.async_tools import async_event
+from ..utils.async_tools import CancelationToken, async_event
 from ..utils.dataclasses import as_json, from_dict
 from ..utils.inspect import ensure_coroutine, iter_methods
 from ..utils.logging import LoggingDescriptor
@@ -57,8 +57,8 @@
     "JsonRPCErrorException",
 ]
 
-T = TypeVar("T")
-TResult = TypeVar("TResult", bound=Any)
+_T = TypeVar("_T")
+_TResult = TypeVar("_TResult", bound=Any)
 
 
 class JsonRPCErrors:
@@ -318,6 +318,7 @@ class SendedRequestEntry(NamedTuple):
 class ReceivedRequestEntry(NamedTuple):
     future: asyncio.Future[Any]
     request: Optional[Any]
+    cancel_token: CancelationToken
 
 
 class JsonRPCProtocolBase(asyncio.Protocol, ABC):
@@ -517,10 +518,10 @@ def send_request(
         self,
         method: str,
         params: Optional[Any] = None,
-        return_type_or_converter: Optional[Type[TResult]] = None,
-    ) -> asyncio.Future[TResult]:
+        return_type_or_converter: Optional[Type[_TResult]] = None,
+    ) -> asyncio.Future[_TResult]:
 
-        result: asyncio.Future[TResult] = asyncio.get_event_loop().create_future()
+        result: asyncio.Future[_TResult] = asyncio.get_event_loop().create_future()
 
         with self._sended_request_lock:
             self._sended_request_count += 1
@@ -536,8 +537,8 @@ async def send_request_async(
         self,
         method: str,
         params: Optional[Any] = None,
-        return_type: Optional[Type[TResult]] = None,
-    ) -> TResult:
+        return_type: Optional[Type[_TResult]] = None,
+    ) -> _TResult:
         return await self.send_request(method, params, return_type)
 
     def send_notification(self, method: str, params: Any) -> None:
@@ -629,7 +630,7 @@ def _convert_params(
                 kw_args["params"] = converted_params
         return args, kw_args
 
-    async def handle_request(self, message: JsonRPCRequest) -> None:
+    async def handle_request(self, message: JsonRPCRequest) -> Optional[asyncio.Task[_T]]:
         e = self.registry.get_entry(message.method)
 
         if e is None or not callable(e.method):
@@ -638,14 +639,17 @@ async def handle_request(self, message: JsonRPCRequest) -> None:
                 f"Unknown method: {message.method}",
                 id=message.id,
             )
-            return
+            return None
 
         params = self._convert_params(e.method, e.param_type, message.params)
+        cancel_token = CancelationToken()
 
-        result = asyncio.create_task(ensure_coroutine(e.method)(*params[0], **params[1]), name=message.method)
+        task = asyncio.create_task(
+            ensure_coroutine(e.method)(*params[0], cancel_token=cancel_token, **params[1]), name=message.method
+        )
 
         with self._received_request_lock:
-            self._received_request[message.id] = ReceivedRequestEntry(result, message)
+            self._received_request[message.id] = ReceivedRequestEntry(task, message, cancel_token)
 
         def done(t: asyncio.Task[Any]) -> None:
             try:
@@ -664,18 +668,22 @@ def done(t: asyncio.Task[Any]) -> None:
                 with self._received_request_lock:
                     self._received_request.pop(message.id, None)
 
-        result.add_done_callback(done)
+        task.add_done_callback(done)
+
+        return task
 
     async def cancel_request(self, id: Union[int, str, None]) -> None:
         with self._received_request_lock:
             entry = self._received_request.get(id, None)
         if entry is not None and entry.future is not None and not entry.future.cancelled():
             self._logger.debug(f"try to cancel request {entry.request}")
+            entry.cancel_token.cancel()
             entry.future.cancel()
 
     async def cancel_all_received_request(self) -> None:
         for entry in self._received_request.values():
             if entry is not None and entry.future is not None and not entry.future.cancelled():
+                entry.cancel_token.cancel()
                 entry.future.cancel()
 
     async def handle_notification(self, message: JsonRPCNotification) -> None:

robotcode/language_server/common/parts/completion.py

@@ -116,6 +116,8 @@ async def _text_document_completion(
     async def _completion_item_resolve(
         self,
         params: CompletionItem,
+        *args: Any,
+        **kwargs: Any,
     ) -> CompletionItem:
 
         results: List[CompletionItem] = []

robotcode/language_server/common/parts/hover.py

@@ -4,7 +4,7 @@
 from typing import TYPE_CHECKING, Any, List, Optional
 
 from ....jsonrpc2.protocol import rpc_method
-from ....utils.async_tools import async_tasking_event
+from ....utils.async_tools import CancelationToken, async_tasking_event
 from ....utils.logging import LoggingDescriptor
 from ..has_extend_capabilities import HasExtendCapabilities
 from ..language import language_id_filter
@@ -31,7 +31,9 @@ def __init__(self, parent: LanguageServerProtocol) -> None:
         super().__init__(parent)
 
     @async_tasking_event
-    async def collect(sender, document: TextDocument, position: Position) -> Optional[Hover]:  # NOSONAR
+    async def collect(
+        sender, document: TextDocument, position: Position, cancel_token: Optional[CancelationToken] = None
+    ) -> Optional[Hover]:  # NOSONAR
         ...
 
     def extend_capabilities(self, capabilities: ServerCapabilities) -> None:
@@ -40,7 +42,12 @@ def extend_capabilities(self, capabilities: ServerCapabilities) -> None:
 
     @rpc_method(name="textDocument/hover", param_type=HoverParams)
     async def _text_document_hover(
-        self, text_document: TextDocumentIdentifier, position: Position, *args: Any, **kwargs: Any
+        self,
+        text_document: TextDocumentIdentifier,
+        position: Position,
+        cancel_token: Optional[CancelationToken] = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> Optional[Hover]:
 
         results: List[Hover] = []
@@ -49,6 +56,7 @@ async def _text_document_hover(
             self,
             document,
             position,
+            cancel_token=cancel_token,
             callback_filter=language_id_filter(document),
         ):
             if isinstance(result, BaseException):

robotcode/language_server/common/parts/workspace.py

@@ -125,7 +125,7 @@ class ConfigBase(Model):
     pass
 
 
-_TConfig = TypeVar("_TConfig", bound=(ConfigBase))
+_TConfig = TypeVar("_TConfig", bound=ConfigBase)
 _F = TypeVar("_F", bound=Callable[..., Any])
 
 

robotcode/language_server/common/protocol.py

@@ -152,6 +152,7 @@ async def _initialize(
         trace: Optional[TraceValue] = None,
         client_info: Optional[ClientInfo] = None,
         workspace_folders: Optional[List[WorkspaceFolder]] = None,
+        *args: Any,
         **kwargs: Any,
     ) -> InitializeResult:
 
@@ -188,7 +189,7 @@ async def on_initialize(sender, initialization_options: Optional[Any] = None) ->
         ...
 
     @rpc_method(name="initialized", param_type=InitializedParams)
-    async def _initialized(self, params: InitializedParams) -> None:
+    async def _initialized(self, params: InitializedParams, *args: Any, **kwargs: Any) -> None:
         await self.on_initialized(self)
 
     @async_event
@@ -197,7 +198,7 @@ async def on_initialized(sender) -> None:  # pragma: no cover, NOSONAR
 
     @rpc_method(name="shutdown")
     @_logger.call
-    async def shutdown(self) -> None:
+    async def shutdown(self, *args: Any, **kwargs: Any) -> None:
         self.shutdown_received = True
 
         try:
@@ -211,12 +212,12 @@ async def shutdown(self) -> None:
 
     @rpc_method(name="exit")
     @_logger.call
-    async def _exit(self) -> None:
+    async def _exit(self, *args: Any, **kwargs: Any) -> None:
         raise SystemExit(0 if self.shutdown_received else 1)
 
     @rpc_method(name="$/setTrace", param_type=SetTraceParams)
     @_logger.call
-    async def _set_trace(self, value: TraceValue, **kwargs: Any) -> None:
+    async def _set_trace(self, value: TraceValue, *args: Any, **kwargs: Any) -> None:
         self.trace = value
 
     @rpc_method(name="$/cancelRequest", param_type=CancelParams)

robotcode/language_server/robotframework/diagnostics/analyzer.py

@@ -41,7 +41,7 @@ async def get(
 
     async def visit(self, node: ast.AST) -> None:
         if self.cancelation_token:
-            self.cancelation_token.throw_if_canceled()
+            self.cancelation_token.raise_if_canceled()
 
         await super().visit(node)