Merge #7236

7236: Fix progress token is already registered crash r=matklad a=matklad

bors r+
🤖

Co-authored-by: Aleksey Kladov <aleksey.kladov@gmail.com>

crates/rust-analyzer/src/global_state.rs

@@ -22,6 +22,7 @@
     from_proto,
     line_endings::LineEndings,
     main_loop::Task,
+    op_queue::OpQueue,
     reload::SourceRootConfig,
     request_metrics::{LatestRequests, RequestMetrics},
     thread_pool::TaskPool,
@@ -78,6 +79,7 @@ pub(crate) struct GlobalState {
     pub(crate) source_root_config: SourceRootConfig,
     pub(crate) proc_macro_client: Option<ProcMacroClient>,
     pub(crate) workspaces: Arc<Vec<ProjectWorkspace>>,
+    pub(crate) fetch_workspaces_queue: OpQueue,
     latest_requests: Arc<RwLock<LatestRequests>>,
 }
 
@@ -130,6 +132,7 @@ pub(crate) fn new(sender: Sender<lsp_server::Message>, config: Config) -> Global
             source_root_config: SourceRootConfig::default(),
             proc_macro_client: None,
             workspaces: Arc::new(Vec::new()),
+            fetch_workspaces_queue: OpQueue::default(),
             latest_requests: Default::default(),
         }
     }

crates/rust-analyzer/src/lib.rs

@@ -35,6 +35,7 @@ macro_rules! eprintln {
 mod thread_pool;
 mod document;
 mod diff;
+mod op_queue;
 pub mod lsp_ext;
 pub mod config;
 

crates/rust-analyzer/src/main_loop.rs

@@ -11,7 +11,6 @@
 use ide_db::base_db::VfsPath;
 use lsp_server::{Connection, Notification, Request, Response};
 use lsp_types::notification::Notification as _;
-use project_model::ProjectWorkspace;
 use vfs::ChangeKind;
 
 use crate::{
@@ -62,7 +61,6 @@ enum Event {
 pub(crate) enum Task {
     Response(Response),
     Diagnostics(Vec<(FileId, Vec<lsp_types::Diagnostic>)>),
-    Workspaces(Vec<anyhow::Result<ProjectWorkspace>>),
     PrimeCaches(PrimeCachesProgress),
     FetchWorkspace(ProjectWorkspaceProgress),
 }
@@ -143,7 +141,8 @@ fn run(mut self, inbox: Receiver<lsp_server::Message>) -> Result<()> {
             |_, _| (),
         );
 
-        self.fetch_workspaces();
+        self.fetch_workspaces_request();
+        self.fetch_workspaces_if_needed();
 
         while let Some(event) = self.next_event(&inbox) {
             if let Event::Lsp(lsp_server::Message::Notification(not)) = &event {
@@ -204,7 +203,6 @@ fn handle_event(&mut self, event: Event) -> Result<()> {
                                 self.diagnostics.set_native_diagnostics(file_id, diagnostics)
                             }
                         }
-                        Task::Workspaces(workspaces) => self.switch_workspaces(workspaces),
                         Task::PrimeCaches(progress) => match progress {
                             PrimeCachesProgress::Started => prime_caches_progress.push(progress),
                             PrimeCachesProgress::StartedOnCrate { .. } => {
@@ -224,7 +222,11 @@ fn handle_event(&mut self, event: Event) -> Result<()> {
                                 ProjectWorkspaceProgress::Report(msg) => {
                                     (Progress::Report, Some(msg))
                                 }
-                                ProjectWorkspaceProgress::End => (Progress::End, None),
+                                ProjectWorkspaceProgress::End(workspaces) => {
+                                    self.fetch_workspaces_completed();
+                                    self.switch_workspaces(workspaces);
+                                    (Progress::End, None)
+                                }
                             };
                             self.report_progress("fetching", state, msg, None);
                         }
@@ -403,6 +405,8 @@ fn handle_event(&mut self, event: Event) -> Result<()> {
             }
         }
 
+        self.fetch_workspaces_if_needed();
+
         let loop_duration = loop_start.elapsed();
         if loop_duration > Duration::from_millis(100) {
             log::warn!("overly long loop turn: {:?}", loop_duration);
@@ -440,7 +444,7 @@ fn on_request(&mut self, request_received: Instant, req: Request) -> Result<()>
         }
 
         RequestDispatcher { req: Some(req), global_state: self }
-            .on_sync::<lsp_ext::ReloadWorkspace>(|s, ()| Ok(s.fetch_workspaces()))?
+            .on_sync::<lsp_ext::ReloadWorkspace>(|s, ()| Ok(s.fetch_workspaces_request()))?
             .on_sync::<lsp_ext::JoinLines>(|s, p| handlers::handle_join_lines(s.snapshot(), p))?
             .on_sync::<lsp_ext::OnEnter>(|s, p| handlers::handle_on_enter(s.snapshot(), p))?
             .on_sync::<lsp_types::request::Shutdown>(|s, ()| {

crates/rust-analyzer/src/op_queue.rs

@@ -0,0 +1,25 @@
+//! Bookkeeping to make sure only one long-running operation is executed.
+
+#[derive(Default)]
+pub(crate) struct OpQueue {
+    op_scheduled: bool,
+    op_in_progress: bool,
+}
+
+impl OpQueue {
+    pub(crate) fn request_op(&mut self) {
+        self.op_scheduled = true;
+    }
+    pub(crate) fn should_start_op(&mut self) -> bool {
+        if !self.op_in_progress && self.op_scheduled {
+            self.op_in_progress = true;
+            self.op_scheduled = false;
+            return true;
+        }
+        false
+    }
+    pub(crate) fn op_completed(&mut self) {
+        assert!(self.op_in_progress);
+        self.op_in_progress = false;
+    }
+}

crates/rust-analyzer/src/reload.rs

@@ -19,7 +19,7 @@
 pub(crate) enum ProjectWorkspaceProgress {
     Begin,
     Report(String),
-    End,
+    End(Vec<anyhow::Result<ProjectWorkspace>>),
 }
 
 impl GlobalState {
@@ -30,7 +30,7 @@ pub(crate) fn update_configuration(&mut self, config: Config) {
             self.analysis_host.update_lru_capacity(self.config.lru_capacity());
         }
         if self.config.linked_projects() != old_config.linked_projects() {
-            self.fetch_workspaces()
+            self.fetch_workspaces_request()
         } else if self.config.flycheck() != old_config.flycheck() {
             self.reload_flycheck();
         }
@@ -44,7 +44,7 @@ pub(crate) fn maybe_refresh(&mut self, changes: &[(AbsPathBuf, ChangeKind)]) {
             Status::Ready | Status::Invalid => (),
         }
         if self.config.cargo_autoreload() {
-            self.fetch_workspaces();
+            self.fetch_workspaces_request();
         } else {
             self.transition(Status::NeedsReload);
         }
@@ -98,8 +98,15 @@ pub(crate) fn transition(&mut self, new_status: Status) {
             });
         }
     }
-    pub(crate) fn fetch_workspaces(&mut self) {
+
+    pub(crate) fn fetch_workspaces_request(&mut self) {
+        self.fetch_workspaces_queue.request_op()
+    }
+    pub(crate) fn fetch_workspaces_if_needed(&mut self) {
         log::info!("will fetch workspaces");
+        if !self.fetch_workspaces_queue.should_start_op() {
+            return;
+        }
 
         self.task_pool.handle.spawn_with_sender({
             let linked_projects = self.config.linked_projects();
@@ -133,12 +140,17 @@ pub(crate) fn fetch_workspaces(&mut self) {
                     })
                     .collect::<Vec<_>>();
 
-                sender.send(Task::FetchWorkspace(ProjectWorkspaceProgress::End)).unwrap();
                 log::info!("did fetch workspaces {:?}", workspaces);
-                sender.send(Task::Workspaces(workspaces)).unwrap()
+                sender
+                    .send(Task::FetchWorkspace(ProjectWorkspaceProgress::End(workspaces)))
+                    .unwrap();
             }
         });
     }
+    pub(crate) fn fetch_workspaces_completed(&mut self) {
+        self.fetch_workspaces_queue.op_completed()
+    }
+
     pub(crate) fn switch_workspaces(&mut self, workspaces: Vec<anyhow::Result<ProjectWorkspace>>) {
         let _p = profile::span("GlobalState::switch_workspaces");
         log::info!("will switch workspaces: {:?}", workspaces);