perf: massive overall speed improvements

Mainly from changing locks from async.Lock to threading.Lock.
Extra: more timing statistics in log output

Author: d-biehl

robotcode/language_server/common/parts/diagnostics.py

@@ -121,8 +121,14 @@ def __init__(self, protocol: LanguageServerProtocol) -> None:
         self._workspace_diagnostics_task: Optional[asyncio.Task[Any]] = None
 
         self._diagnostics_loop: Optional[asyncio.AbstractEventLoop] = None
+        self._single_diagnostics_loop: Optional[asyncio.AbstractEventLoop] = None
+
         self._diagnostics_loop_lock = threading.RLock()
         self._diagnostics_started = threading.Event()
+        self._single_diagnostics_started = threading.Event()
+
+        self._diagnostics_server_thread: Optional[threading.Thread] = None
+        self._single_diagnostics_server_thread: Optional[threading.Thread] = None
 
         self.parent.on_initialized.add(self.initialized)
 
@@ -148,6 +154,15 @@ def diagnostics_loop(self) -> asyncio.AbstractEventLoop:
 
         return self._diagnostics_loop
 
+    @property
+    def single_diagnostics_loop(self) -> asyncio.AbstractEventLoop:
+        if self._single_diagnostics_loop is None:
+            self._ensure_diagnostics_thread_started()
+
+        assert self._single_diagnostics_loop is not None
+
+        return self._single_diagnostics_loop
+
     def _run_diagnostics(self) -> None:
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
@@ -164,17 +179,39 @@ def _run_diagnostics(self) -> None:
             asyncio.set_event_loop(None)
             loop.close()
 
+    def _single_run_diagnostics(self) -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            self._single_diagnostics_loop = loop
+            self._single_diagnostics_started.set()
+
+            loop.slow_callback_duration = 10
+
+            loop.run_forever()
+            _cancel_all_tasks(loop)
+            loop.run_until_complete(loop.shutdown_asyncgens())
+        finally:
+            asyncio.set_event_loop(None)
+            loop.close()
+
     def _ensure_diagnostics_thread_started(self) -> None:
         with self._diagnostics_loop_lock:
             if self._diagnostics_loop is None:
-                self._server_thread = threading.Thread(
+                self._diagnostics_server_thread = threading.Thread(
                     name="diagnostics_worker", target=self._run_diagnostics, daemon=True
                 )
 
-                self._server_thread.start()
+                self._diagnostics_server_thread.start()
+
+                self._single_diagnostics_server_thread = threading.Thread(
+                    name="single_diagnostics_worker", target=self._single_run_diagnostics, daemon=True
+                )
+
+                self._single_diagnostics_server_thread.start()
 
-                if not self._diagnostics_started.wait(10):
-                    raise RuntimeError("Can't start diagnostics worker thread.")
+                if not self._diagnostics_started.wait(10) or not self._single_diagnostics_started.wait(10):
+                    raise RuntimeError("Can't start diagnostics worker threads.")
 
     def extend_capabilities(self, capabilities: ServerCapabilities) -> None:
         if (
@@ -280,7 +317,6 @@ async def run_workspace_diagnostics(self) -> None:
 
         while True:
             try:
-
                 documents = [
                     doc
                     for doc in self.parent.documents.documents
@@ -296,8 +332,11 @@ async def run_workspace_diagnostics(self) -> None:
                     await asyncio.sleep(1)
                     continue
 
+                self._logger.info(f"start collecting workspace diagnostics for {len(documents)} documents")
+
                 done_something = False
 
+                start = time.monotonic()
                 async with self.parent.window.progress(
                     "Analyse workspace", cancellable=False, current=0, max=len(documents) + 1, start=False
                 ) as progress:
@@ -324,21 +363,31 @@ async def run_workspace_diagnostics(self) -> None:
                             progress.report(current=i + 1)
 
                         await self.create_document_diagnostics_task(
-                            document, False, await self.get_diagnostics_mode(document.uri) == DiagnosticsMode.WORKSPACE
+                            document,
+                            False,
+                            False,
+                            await self.get_diagnostics_mode(document.uri) == DiagnosticsMode.WORKSPACE,
                         )
 
                 if not done_something:
                     await asyncio.sleep(1)
+
+                self._logger.info(
+                    f"collecting workspace diagnostics for for {len(documents)} "
+                    f"documents takes {time.monotonic() - start}s"
+                )
+
             except (SystemExit, KeyboardInterrupt, asyncio.CancelledError):
                 raise
             except BaseException as e:
                 self._logger.exception(f"Error in workspace diagnostics loop: {e}", exc_info=e)
 
     def create_document_diagnostics_task(
-        self, document: TextDocument, debounce: bool = True, send_diagnostics: bool = True
+        self, document: TextDocument, single: bool, debounce: bool = True, send_diagnostics: bool = True
     ) -> asyncio.Task[Any]:
         def done(t: asyncio.Task[Any]) -> None:
             self._logger.debug(lambda: f"diagnostics for {document} {'canceled' if t.cancelled() else 'ended'}")
+
             if t.done() and not t.cancelled():
                 ex = t.exception()
 
@@ -372,7 +421,7 @@ async def cancel(t: asyncio.Task[Any]) -> None:
             data.version = document.version
             data.task = create_sub_task(
                 self._get_diagnostics_for_document(document, data, debounce, send_diagnostics),
-                loop=self.diagnostics_loop,
+                loop=self.single_diagnostics_loop if single else self.diagnostics_loop,
                 name=f"diagnostics ${document.uri}",
             )
 
@@ -444,7 +493,7 @@ async def _text_document_diagnostic(
             if document is None:
                 raise JsonRPCErrorException(ErrorCodes.SERVER_CANCELLED, f"Document {text_document!r} not found.")
 
-            self.create_document_diagnostics_task(document)
+            self.create_document_diagnostics_task(document, True)
 
             return RelatedFullDocumentDiagnosticReport([])
         except asyncio.CancelledError:

robotcode/language_server/common/parts/documents.py

@@ -111,7 +111,7 @@ async def _file_watcher(self, sender: Any, changes: List[FileEvent]) -> None:
                         loop=self.parent.loop,
                     )
                 elif change.type == FileChangeType.CHANGED:
-                    await document.apply_full_change(
+                    document.apply_full_change(
                         None, await self.read_document_text(document.uri, language_id_filter(document)), save=True
                     )
                     create_sub_task(
@@ -258,9 +258,9 @@ async def _text_document_did_open(self, text_document: TextDocumentItem, *args:
 
                 self._documents[uri] = document
             else:
-                text_changed = await document.text() != normalized_text
+                text_changed = document.text() != normalized_text
                 if text_changed:
-                    await document.apply_full_change(text_document.version, normalized_text)
+                    document.apply_full_change(text_document.version, normalized_text)
 
             document.opened_in_editor = True
 
@@ -297,11 +297,11 @@ async def close_document(self, document: TextDocument, real_close: bool = False)
             async with self._lock:
                 self._documents.pop(str(document.uri), None)
 
-            await document.clear()
+            document.clear()
             del document
         else:
             document._version = None
-            if await document.revert(None):
+            if document.revert(None):
                 create_sub_task(
                     self.did_change(self, document, callback_filter=language_id_filter(document)), loop=self.parent.loop
                 )
@@ -326,9 +326,9 @@ async def _text_document_did_save(
             if text is not None:
                 normalized_text = self._normalize_line_endings(text)
 
-                text_changed = await document.text() != normalized_text
+                text_changed = document.text() != normalized_text
                 if text_changed:
-                    await document.save(None, text)
+                    document.save(None, text)
                     create_sub_task(
                         self.did_change(self, document, callback_filter=language_id_filter(document)),
                         loop=self.parent.loop,
@@ -367,17 +367,15 @@ async def _text_document_did_change(
         )
         for content_change in content_changes:
             if sync_kind is None or sync_kind == TextDocumentSyncKind.NONE:
-                await document.apply_none_change()
+                document.apply_none_change()
             elif sync_kind == TextDocumentSyncKind.FULL and isinstance(
                 content_change, TextDocumentContentTextChangeEvent
             ):
-                await document.apply_full_change(
-                    text_document.version, self._normalize_line_endings(content_change.text)
-                )
+                document.apply_full_change(text_document.version, self._normalize_line_endings(content_change.text))
             elif sync_kind == TextDocumentSyncKind.INCREMENTAL and isinstance(
                 content_change, TextDocumentContentRangeChangeEvent
             ):
-                await document.apply_incremental_change(
+                document.apply_incremental_change(
                     text_document.version, content_change.range, self._normalize_line_endings(content_change.text)
                 )
             else:

robotcode/language_server/common/parts/workspace.py

@@ -24,12 +24,7 @@
 )
 
 from ....jsonrpc2.protocol import rpc_method
-from ....utils.async_tools import (
-    Lock,
-    async_event,
-    async_tasking_event,
-    create_sub_task,
-)
+from ....utils.async_tools import Lock, async_tasking_event, create_sub_task, threaded
 from ....utils.dataclasses import from_dict
 from ....utils.logging import LoggingDescriptor
 from ....utils.path import path_is_relative_to
@@ -388,11 +383,12 @@ async def _workspace_did_change_workspace_folders(
             for a in event.added:
                 self._workspace_folders.append(WorkspaceFolder(a.name, Uri(a.uri), a.uri))
 
-    @async_event
+    @async_tasking_event
     async def did_change_watched_files(sender, changes: List[FileEvent]) -> None:  # NOSONAR
         ...
 
     @rpc_method(name="workspace/didChangeWatchedFiles", param_type=DidChangeWatchedFilesParams)
+    @threaded()
     async def _workspace_did_change_watched_files(self, changes: List[FileEvent], *args: Any, **kwargs: Any) -> None:
         changes = [e for e in changes if not e.uri.endswith("/globalStorage")]
         if changes: