optimize locking and get tests from workspace

Author: d-biehl

CHANGELOG.md

@@ -717,6 +717,12 @@ async def handle_request(self, message: JsonRPCRequest) -> None:
 
         def done(t: asyncio.Task[Any]) -> None:
             try:
+                if not t.cancelled():
+                    ex = t.exception()
+                    if ex is not None:
+                        self._logger.exception(ex, exc_info=ex)
+                        raise JsonRPCErrorException(JsonRPCErrors.INTERNAL_ERROR, f"{type(ex).__name__}: {ex}") from ex
+
                 self.send_response(message.id, t.result())
             except asyncio.CancelledError:
                 self._logger.debug(f"request message {repr(message)} canceled")

robotcode/jsonrpc2/protocol.py

@@ -370,7 +370,10 @@ async def cancel(t: asyncio.Task[Any]) -> None:
                     except asyncio.CancelledError:
                         pass
 
-                asyncio.run_coroutine_threadsafe(cancel(task), loop=task.get_loop()).result(5)
+                try:
+                    asyncio.run_coroutine_threadsafe(cancel(task), loop=task.get_loop()).result(600)
+                except TimeoutError as e:
+                    raise RuntimeError("Can't cancel diagnostics task.") from e
 
                 # task.get_loop().call_soon_threadsafe(task.cancel)
 

robotcode/language_server/common/parts/diagnostics.py

@@ -138,7 +138,7 @@ def calc_md() -> str:
                             name + ("::" + args if args else ""),
                             base_dir=basedir if basedir else ".",
                             theme=theme,
-                        ).result(10)
+                        ).result(600)
 
                         self.send_response(200)
                         self.send_header("Content-type", "text/html")

robotcode/language_server/robotframework/parts/code_action.py

@@ -24,7 +24,6 @@
 )
 
 from ....utils.async_itertools import async_chain, async_chain_iterator
-from ....utils.async_tools import threaded
 from ....utils.logging import LoggingDescriptor
 from ...common.decorators import language_id, trigger_characters
 from ...common.lsp_types import (
@@ -108,7 +107,6 @@ async def get_header_style(self, config: CompletionConfig) -> str:
     )
     # @all_commit_characters(['\n'])
     @language_id("robotframework")
-    @threaded()
     @_logger.call
     async def collect(
         self, sender: Any, document: TextDocument, position: Position, context: Optional[CompletionContext]
@@ -130,7 +128,6 @@ async def collect(
         )
 
     @language_id("robotframework")
-    @threaded()
     @_logger.call
     async def resolve(self, sender: Any, completion_item: CompletionItem) -> CompletionItem:
         if completion_item.data is not None:

robotcode/language_server/robotframework/parts/completion.py

@@ -8,7 +8,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Iterator, List, Optional, cast
 
 from ....jsonrpc2.protocol import rpc_method
-from ....utils.async_tools import run_coroutine_in_thread
+from ....utils.async_tools import threaded
 from ....utils.logging import LoggingDescriptor
 from ....utils.uri import Uri
 from ...common.lsp_types import (
@@ -62,7 +62,7 @@ class FindTestCasesVisitor(AsyncVisitor):
     async def get(self, source: DocumentUri, model: ast.AST, base_name: Optional[str]) -> List[TestItem]:
         self._results: List[TestItem] = []
         self.source = source
-        self.path = Uri(source).to_path().resolve()
+        self.path = Uri(source).to_path()
         self.base_name = base_name
         await self.visit(model)
         return self._results
@@ -111,8 +111,15 @@ async def get_config(self, workspace_uri: str) -> Optional[RobotConfig]:
 
         return await self.parent.workspace.get_configuration(RobotConfig, folder.uri)
 
-    async def _get_tests_from_workspace(
-        self, workspace_folder: str, paths: Optional[List[str]], suites: Optional[List[str]]
+    @rpc_method(name="robot/discovering/getTestsFromWorkspace", param_type=GetAllTestsParams)
+    @threaded()
+    async def get_tests_from_workspace(
+        self,
+        workspace_folder: str,
+        paths: Optional[List[str]],
+        suites: Optional[List[str]],
+        *args: Any,
+        **kwargs: Any,
     ) -> List[TestItem]:
 
         from robot.output.logger import LOGGER
@@ -138,7 +145,7 @@ async def _get_tests_from_workspace(
         def get_document_text(source: str) -> str:
             if self.parent._loop:
                 doc = self.parent.documents.get_sync(Uri.from_path(source).normalized())
-                if doc is not None and doc.opened_in_editor:
+                if doc is not None:
                     return doc.text_sync()
 
             return source
@@ -258,8 +265,7 @@ def generate(suite: TestSuite) -> TestItem:
                 children.append(
                     TestItem(
                         type="test",
-                        id=f"{Path(test.source).resolve() if test.source is not None else ''};"
-                        f"{test.longname};{test.lineno}",
+                        id=f"{test.source if test.source is not None else ''};" f"{test.longname};{test.lineno}",
                         label=test.name,
                         longname=test.longname,
                         uri=str(Uri.from_path(test.source)) if test.source else None,
@@ -276,7 +282,7 @@ def generate(suite: TestSuite) -> TestItem:
 
             return TestItem(
                 type="suite",
-                id=f"{Path(suite.source).resolve() if suite.source is not None else ''};{suite.longname}",
+                id=f"{suite.source if suite.source is not None else ''};{suite.longname}",
                 label=suite.name,
                 longname=suite.longname,
                 uri=str(Uri.from_path(suite.source)) if suite.source else None,
@@ -342,7 +348,7 @@ def nonexisting_paths(paths: List[str]) -> Iterator[str]:
                     return [
                         TestItem(
                             type="workspace",
-                            id=str(Path.cwd().resolve()),
+                            id=str(Path.cwd()),
                             label=Path.cwd().name,
                             longname=Path.cwd().name,
                             uri=str(Uri.from_path(Path.cwd())),
@@ -375,44 +381,30 @@ def nonexisting_paths(paths: List[str]) -> Iterator[str]:
                 return [
                     TestItem(
                         type="error",
-                        id=str(Uri.from_path(Path.cwd().resolve())),
+                        id=str(Uri.from_path(Path.cwd())),
                         longname="error",
                         label=Path.cwd().name,
                         error=str(e),
                     )
                 ]
 
-    @rpc_method(name="robot/discovering/getTestsFromWorkspace", param_type=GetAllTestsParams)
-    @_logger.call
-    async def get_tests_from_workspace(
-        self,
-        workspace_folder: str,
-        paths: Optional[List[str]],
-        suites: Optional[List[str]],
-        *args: Any,
-        **kwargs: Any,
-    ) -> List[TestItem]:
-        return await run_coroutine_in_thread(self._get_tests_from_workspace, workspace_folder, paths, suites)
-
     @rpc_method(name="robot/discovering/getTestsFromDocument", param_type=GetTestsParams)
-    @_logger.call
+    @threaded()
     async def get_tests_from_document(
         self, text_document: TextDocumentIdentifier, base_name: Optional[str], *args: Any, **kwargs: Any
     ) -> List[TestItem]:
-        async def run() -> List[TestItem]:
-            try:
-                return await FindTestCasesVisitor().get(
-                    text_document.uri,
-                    await self.parent.documents_cache.get_model(
-                        await self.parent.documents.get_or_open_document(
-                            Uri(text_document.uri).to_path(), language_id="robotframework"
-                        )
-                    ),
-                    base_name,
-                )
-            except (asyncio.CancelledError, SystemExit, KeyboardInterrupt):
-                raise
-            except BaseException:
-                return []
 
-        return await run_coroutine_in_thread(run)
+        try:
+            return await FindTestCasesVisitor().get(
+                text_document.uri,
+                await self.parent.documents_cache.get_model(
+                    await self.parent.documents.get_or_open_document(
+                        Uri(text_document.uri).to_path(), language_id="robotframework"
+                    )
+                ),
+                base_name,
+            )
+        except (asyncio.CancelledError, SystemExit, KeyboardInterrupt):
+            raise
+        except BaseException:
+            return []

robotcode/language_server/robotframework/parts/discovering.py

@@ -56,6 +56,7 @@ def __init__(self, parent: RobotLanguageServerProtocol) -> None:
     async def cache_cleared(sender) -> None:  # NOSONAR
         ...
 
+    @language_id("robotframework")
     @threaded()
     async def document_did_change(self, sender: Any, document: TextDocument) -> None:
         await self._keyword_reference_cache.clear()

robotcode/language_server/robotframework/parts/references.py

@@ -11,7 +11,6 @@
 import weakref
 from collections import deque
 from concurrent.futures.thread import ThreadPoolExecutor
-from contextlib import asynccontextmanager
 from types import TracebackType
 from typing import (
     Any,
@@ -380,8 +379,8 @@ def __repr__(self) -> str:
         return f"<{res[1:-1]} [{extra}]>"
 
     def is_set(self) -> bool:
-        # with self._lock:
-        return self._value
+        with self._lock:
+            return self._value
 
     def set(self) -> None:
         with self._lock:
@@ -443,15 +442,15 @@ class Lock:
     def __init__(self) -> None:
         self._waiters: Optional[Deque[asyncio.Future[Any]]] = None
         self._locked = False
-        self._lock = threading.Lock()
+        self._lock = threading.RLock()
 
     async def __aenter__(self) -> None:
         await self.acquire()
 
     async def __aexit__(
         self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
     ) -> None:
-        await self.release()
+        self.release()
 
     def __repr__(self) -> str:
         res = super().__repr__()
@@ -460,24 +459,12 @@ def __repr__(self) -> str:
             extra = f"{extra}, waiters:{len(self._waiters)}"
         return f"<{res[1:-1]} [{extra}]>"
 
-    @asynccontextmanager
-    async def __inner_lock(self) -> AsyncGenerator[Any, None]:
-        # while not (b := self._lock.acquire(blocking=False)):
-        #     await asyncio.sleep(0)
-        # try:
-        #     yield None
-        # finally:
-        #     if b:
-        #         self._lock.release()
-        with self._lock:
-            yield None
-
     @property
     def locked(self) -> bool:
         return self._locked
 
     async def acquire(self) -> bool:
-        async with self.__inner_lock():
+        with self._lock:
             if not self._locked and (self._waiters is None or all(w.cancelled() for w in self._waiters)):
                 self._locked = True
                 return True
@@ -489,42 +476,49 @@ async def acquire(self) -> bool:
             self._waiters.append(fut)
 
         try:
+            try:
 
-            def aaa() -> None:
-                warnings.warn(f"Lock takes to long {threading.current_thread()}")
+                def aaa() -> None:
+                    warnings.warn(f"Lock takes to long {threading.current_thread()}")
 
-            h = fut.get_loop().call_later(320, aaa)
+                h = fut.get_loop().call_later(60, aaa)
 
-            await fut
+                await fut
 
-            h.cancel()
-        finally:
-            async with self.__inner_lock():
-                if fut in self._waiters:
+                h.cancel()
+            finally:
+                with self._lock:
                     self._waiters.remove(fut)
-                self._locked = True
+        except asyncio.CancelledError:
+            with self._lock:
+                if not self._locked:
+                    self._wake_up_first()
+                raise
+
+        with self._lock:
+            self._locked = True
 
         return True
 
-    async def release(self) -> None:
-        async with self.__inner_lock():
-            if self._waiters is None or len(self._waiters) == 0:
-                if self._locked:
-                    self._locked = False
+    def release(self) -> None:
+        with self._lock:
+            wake_up = False
+            if self._locked:
+                self._locked = False
+                wake_up = True
 
-        await self._wake_up_next()
+        if wake_up:
+            self._wake_up_first()
 
-    async def _wake_up_next(self) -> None:
+    def _wake_up_first(self) -> None:
         if not self._waiters:
             return
 
-        async with self.__inner_lock():
+        with self._lock:
             try:
                 fut = next(iter(self._waiters))
             except StopIteration:
                 return
-            if fut in self._waiters:
-                self._waiters.remove(fut)
 
         if fut.get_loop().is_running() and not fut.get_loop().is_closed():
             if fut.get_loop() == asyncio.get_running_loop():
@@ -551,10 +545,12 @@ def set_result(w: asyncio.Future[Any], ev: threading.Event) -> None:
                             warnings.warn("Can't set future result.")
                             break
 
-                        await asyncio.sleep(0)
+                        time.sleep(0.001)
         else:
             warnings.warn(f"Future {repr(fut)} loop is closed")
-            await self._wake_up_next()
+            with self._lock:
+                self._waiters.remove(fut)
+            self._wake_up_first()
 
 
 class FutureInfo: