refactor: isolate qihe actor

src/global_state.rs

@@ -27,7 +27,7 @@ use hir::base_db::{
 };
 use ide::analysis_host::AnalysisHost;
 use lsp_server::{Message, ReqQueue, Request};
-use lsp_types::{TraceValue, Url};
+use lsp_types::{NumberOrString, TraceValue, Url};
 use nohash_hasher::IntMap;
 use parking_lot::{Mutex, RwLock};
 use project_model::Workspace;
@@ -48,11 +48,14 @@ use self::{
     },
     mem_docs::MemDocs,
     snapshot::GlobalStateSnapshot,
-    task::{Task, TaskPool},
+    task::{QiheTask, Task, TaskPool},
     trace::LspTrace,
     workspace_state::WorkspaceVfsReadiness,
 };
-use crate::config::{Config, ConfigError};
+use crate::{
+    config::{Config, ConfigError},
+    lsp_ext::ext::RunQiheAnalysisParams,
+};
 
 pub(crate) struct Handle<H, C> {
     pub(crate) handle: H,
@@ -68,7 +71,7 @@ pub(crate) struct GlobalState {
     pub(crate) analysis: AnalysisState,
     pub(crate) diagnostics: DiagnosticsState,
     pub(crate) workspace: WorkspaceState,
-    pub(crate) qihe: QiheState,
+    pub(crate) qihe: qihe::Qihe,
     pub(crate) tasks: TaskState,
 }
 
@@ -107,14 +110,6 @@ pub(crate) struct DiagnosticsState {
     pub(crate) diagnostic_file_revisions: FxHashMap<FileId, DiagnosticFileRevision>,
 }
 
-pub(crate) struct QiheState {
-    pub(crate) qihe_diagnostics: Arc<Mutex<FxHashMap<FileId, QiheDiagnosticState>>>,
-    // Only the latest Qihe run is allowed to commit diagnostics or logs.
-    pub(crate) qihe_run_generation: qihe::QiheRunId,
-    pub(crate) qihe_active_progress_token: Option<String>,
-    pub(crate) qihe_active_cancel_token: Option<CancellationToken>,
-}
-
 pub(crate) struct WorkspaceState {
     pub(crate) vfs_loader: Handle<Box<dyn vfs::loader::Handle>, Receiver<vfs::loader::Message>>,
     pub(crate) vfs: Arc<RwLock<(Vfs, IntMap<FileId, LineEnding>)>>,
@@ -186,12 +181,7 @@ impl GlobalState {
                 fetch_workspaces_task: ExclTask::default(),
                 registered_client_file_watcher_globs: None,
             },
-            qihe: QiheState {
-                qihe_diagnostics: Arc::new(Mutex::new(FxHashMap::default())),
-                qihe_run_generation: qihe::QiheRunId::default(),
-                qihe_active_progress_token: None,
-                qihe_active_cancel_token: None,
-            },
+            qihe: qihe::Qihe::new(),
             tasks: TaskState { task_pool },
         }
     }
@@ -211,7 +201,7 @@ impl GlobalState {
             vfs: Arc::clone(&self.workspace.vfs),
             mem_docs: self.analysis.mem_docs.clone(),
             sema_tokens_cache: Arc::clone(&self.analysis.semantic_tokens_cache),
-            qihe_diagnostics: Arc::clone(&self.qihe.qihe_diagnostics),
+            qihe_diagnostics: self.qihe.diagnostics_snapshot(),
             diagnostic_publish_freshness: self.diagnostic_publish_freshness(),
             diagnostic_file_revisions: self.diagnostics.diagnostic_file_revisions.clone(),
             cancellation,
@@ -227,8 +217,28 @@ impl GlobalState {
         )
     }
 
-    pub(crate) fn diagnostic_commit_freshness(&self) -> DiagnosticCommitFreshness {
-        self.diagnostic_publish_freshness().commit()
+    pub(crate) fn spawn_qihe_analysis(&mut self, params: RunQiheAnalysisParams) {
+        qihe::with_global_ctx(self, |qihe, ctx| qihe.start(params, ctx));
+    }
+
+    pub(crate) fn handle_qihe_task(&mut self, task: QiheTask) {
+        qihe::with_global_ctx(self, |qihe, ctx| qihe.handle(task, ctx));
+    }
+
+    pub(crate) fn cancel_work_done_progress(
+        &mut self,
+        params: lsp_types::WorkDoneProgressCancelParams,
+    ) {
+        let token = match params.token {
+            NumberOrString::String(token) => token,
+            NumberOrString::Number(token) => token.to_string(),
+        };
+        self.qihe.cancel_progress_token(&token);
+    }
+
+    #[cfg(test)]
+    pub(crate) fn publish_qihe_diagnostics(&mut self, changed_files: FxHashSet<FileId>) {
+        qihe::with_global_ctx(self, |qihe, ctx| qihe.publish_diagnostics(changed_files, ctx));
     }
 }
 

src/global_state/diagnostics/publisher.rs

@@ -1,8 +1,13 @@
-use rustc_hash::FxHashSet;
+use rustc_hash::{FxHashMap, FxHashSet};
 use vfs::FileId;
 
 use super::DiagnosticPublishFreshness;
-use crate::global_state::{GlobalState, snapshot::DiagnosticPublishTarget};
+use crate::{
+    config::Config,
+    global_state::{
+        GlobalState, snapshot::DiagnosticPublishTarget, workspace_state::WorkspaceVfsReadiness,
+    },
+};
 
 #[derive(Debug)]
 pub(crate) struct PublishDiagnosticsTask {
@@ -103,23 +108,38 @@ impl PublishDiagnosticsTask {
     }
 }
 
-impl GlobalState {
-    pub(in crate::global_state) fn publish_diagnostics_tasks(
-        &mut self,
-        batch: PublishDiagnosticsBatch,
-    ) {
+pub(in crate::global_state) struct DiagnosticsPublisher<'a> {
+    config: &'a Config,
+    workspace_vfs: &'a mut WorkspaceVfsReadiness,
+    published_diagnostics: &'a mut FxHashMap<DiagnosticPublishKey, Vec<lsp_types::Diagnostic>>,
+    sender: &'a crossbeam_channel::Sender<lsp_server::Message>,
+    current_freshness: DiagnosticPublishFreshness,
+}
+
+impl<'a> DiagnosticsPublisher<'a> {
+    pub(in crate::global_state) fn new(
+        config: &'a Config,
+        workspace_vfs: &'a mut WorkspaceVfsReadiness,
+        published_diagnostics: &'a mut FxHashMap<DiagnosticPublishKey, Vec<lsp_types::Diagnostic>>,
+        sender: &'a crossbeam_channel::Sender<lsp_server::Message>,
+        current_freshness: DiagnosticPublishFreshness,
+    ) -> Self {
+        Self { config, workspace_vfs, published_diagnostics, sender, current_freshness }
+    }
+
+    pub(in crate::global_state) fn publish(&mut self, batch: PublishDiagnosticsBatch) {
         let task_count = batch.touched_file_count();
         let diagnostic_count = batch.diagnostic_count();
         let _span =
             tracing::info_span!("diagnostics.publish", task_count, diagnostic_count).entered();
 
-        if self.config_state.config.cli_pull_diagnostics_support() {
+        if self.config.cli_pull_diagnostics_support() {
             tracing::info!("skipping push diagnostics for pull-capable client");
             return;
         }
 
-        if !self.workspace.workspace_vfs.is_ready() {
-            self.workspace.workspace_vfs.defer_diagnostics_until_ready();
+        if !self.workspace_vfs.is_ready() {
+            self.workspace_vfs.defer_diagnostics_until_ready();
             tracing::debug!("diagnostics publish deferred until workspace/VFS is ready");
             return;
         }
@@ -128,22 +148,24 @@ impl GlobalState {
         let mut published_diagnostics = 0usize;
         let mut skipped_files = 0usize;
         let PublishDiagnosticsBatch { freshness, touched_file_ids, tasks } = batch;
-        let current_freshness = self.diagnostic_publish_freshness();
-        if freshness != current_freshness {
-            tracing::debug!(?freshness, ?current_freshness, "stale diagnostics batch ignored");
+        if freshness != self.current_freshness {
+            tracing::debug!(
+                freshness = ?freshness,
+                current_freshness = ?self.current_freshness,
+                "stale diagnostics batch ignored"
+            );
             return;
         }
         let current_targets =
             tasks.iter().map(PublishDiagnosticsTask::cache_key).collect::<FxHashSet<_>>();
         let stale_targets = self
-            .diagnostics
             .published_diagnostics
             .keys()
             .filter(|key| touched_file_ids.contains(&key.file_id) && !current_targets.contains(key))
             .cloned()
             .collect::<Vec<_>>();
         for key in stale_targets {
-            self.diagnostics.published_diagnostics.remove(&key);
+            self.published_diagnostics.remove(&key);
             self.send_notification::<lsp_types::notification::PublishDiagnostics>(
                 lsp_types::PublishDiagnosticsParams {
                     uri: key.uri,
@@ -157,7 +179,7 @@ impl GlobalState {
         for diag in tasks {
             let file_diagnostics = diag.diagnostics.len();
             let cache_key = diag.cache_key();
-            let should_publish = match self.diagnostics.published_diagnostics.get(&cache_key) {
+            let should_publish = match self.published_diagnostics.get(&cache_key) {
                 Some(prev) => prev != &diag.diagnostics,
                 None => !diag.diagnostics.is_empty(),
             };
@@ -168,9 +190,9 @@ impl GlobalState {
             }
 
             if diag.diagnostics.is_empty() {
-                self.diagnostics.published_diagnostics.remove(&cache_key);
+                self.published_diagnostics.remove(&cache_key);
             } else {
-                self.diagnostics.published_diagnostics.insert(cache_key, diag.diagnostics.clone());
+                self.published_diagnostics.insert(cache_key, diag.diagnostics.clone());
             }
 
             self.send_notification::<lsp_types::notification::PublishDiagnostics>(
@@ -190,4 +212,28 @@ impl GlobalState {
             "publish diagnostics complete"
         );
     }
+
+    fn send_notification<N: lsp_types::notification::Notification>(&self, params: N::Params) {
+        let notif = lsp_server::Notification::new(N::METHOD.to_string(), params);
+        if self.sender.send(notif.into()).is_err() {
+            tracing::debug!("LSP message dropped because client connection is closed");
+        }
+    }
+}
+
+impl GlobalState {
+    pub(in crate::global_state) fn publish_diagnostics_tasks(
+        &mut self,
+        batch: PublishDiagnosticsBatch,
+    ) {
+        let current_freshness = self.diagnostic_publish_freshness();
+        DiagnosticsPublisher::new(
+            &self.config_state.config,
+            &mut self.workspace.workspace_vfs,
+            &mut self.diagnostics.published_diagnostics,
+            &self.client.sender,
+            current_freshness,
+        )
+        .publish(batch);
+    }
 }

src/global_state/process_changes.rs

@@ -117,7 +117,7 @@ impl GlobalState {
         if diagnostic_targets_changed {
             self.diagnostics.diagnostic_target_revision += 1;
         }
-        self.remove_deleted_qihe_diagnostics(&deleted_file_ids);
+        self.qihe.remove_deleted(&deleted_file_ids);
         self.clear_deleted_push_diagnostics(&deleted_push_diagnostics);
         if has_structure_changes {
             self.invalidate_diagnostics(DiagnosticInvalidation::WorkspaceChanged);
@@ -189,17 +189,6 @@ impl GlobalState {
         self.request_diagnostics(file_ids);
     }
 
-    fn remove_deleted_qihe_diagnostics(&mut self, deleted_file_ids: &FxHashSet<FileId>) {
-        if deleted_file_ids.is_empty() {
-            return;
-        }
-
-        let mut qihe_diagnostics = self.qihe.qihe_diagnostics.lock();
-        for file_id in deleted_file_ids {
-            qihe_diagnostics.remove(file_id);
-        }
-    }
-
     fn clear_deleted_push_diagnostics(&mut self, deleted_files: &[(FileId, VfsPath)]) {
         if deleted_files.is_empty() || self.config_state.config.cli_pull_diagnostics_support() {
             return;