batched ingot-wide diagnostics

crates/language-server/Cargo.toml

@@ -48,3 +48,4 @@ futures-concurrency = "7.5.0"
 console-subscriber = "0.2.0"
 vfs = "0.12.0"
 rust-embed = "8.3.0"
+futures-batch = "0.6.1"

crates/language-server/src/backend.rs

@@ -2,13 +2,16 @@ use crate::handlers::request::{handle_goto_definition, handle_hover};
 
 use crate::workspace::SyncableIngotFileContext;
 
+use futures_batch::ChunksTimeoutStreamExt;
+use common::InputDb;
 use fork_stream::StreamExt as _;
+use fxhash::FxHashSet;
 
 use futures::StreamExt;
 use futures_concurrency::prelude::*;
 use lsp_types::TextDocumentItem;
 use salsa::{ParallelDatabase, Snapshot};
-use stream_operators::StreamOps;
+use tokio_stream::wrappers::UnboundedReceiverStream;
 
 use std::sync::Arc;
 use tokio::sync::RwLock;
@@ -60,12 +63,9 @@ impl Backend {
 
         let client = self.client.clone();
         let messaging = self.messaging;
-        // let messaging = self.messaging.clone();
-        // let messaging = messaging.read().await;
 
         let mut initialized_stream = messaging.initialize_stream.fuse();
         let mut shutdown_stream = messaging.shutdown_stream.fuse();
-        // let mut did_close_stream = messaging.did_close_stream.fuse();
         let did_change_watched_files_stream = messaging.did_change_watched_files_stream.fork();
 
         let flat_did_change_watched_files = did_change_watched_files_stream
@@ -89,7 +89,7 @@ impl Backend {
             flat_did_change_watched_files.clone().filter(|change| {
                 let change_type = change.typ;
                 Box::pin(async move { matches!(change_type, lsp_types::FileChangeType::DELETED) })
-            });
+            }).fuse();
 
         let did_open_stream = messaging.did_open_stream.fuse();
         let did_change_stream = messaging.did_change_stream.fuse();
@@ -130,7 +130,19 @@ impl Backend {
             }),
         )
             .merge()
-            .debounce_time(std::time::Duration::from_millis(20))
+            .fuse();
+
+        let (tx_needs_diagnostics, rx_needs_diagnostics) = tokio::sync::mpsc::unbounded_channel();
+
+        let mut diagnostics_stream = 
+            UnboundedReceiverStream::from(rx_needs_diagnostics)
+            .chunks_timeout(100, std::time::Duration::from_millis(100))
+            .map(|changed_paths| {
+                changed_paths.iter().fold(FxHashSet::default(), |mut acc, path: &String| {
+                    acc.insert(path.clone());
+                    acc
+                })
+            })
             .fuse();
 
         let mut hover_stream = messaging.hover_stream.fuse();
@@ -142,8 +154,6 @@ impl Backend {
                 Some(result) = initialized_stream.next() => {
                     let (initialization_params, responder) = result;
                     info!("initializing language server!");
-                    // setup workspace
-                    // let workspace = self.workspace.clone();
 
                     let root =
                         initialization_params
@@ -182,25 +192,45 @@ impl Backend {
                     let path = path.to_str().unwrap();
                     let workspace = workspace.clone();
                     let _ = workspace.write().await.remove_input_for_file_path(db, path);
+                    let _ = tx_needs_diagnostics.send(path.to_string());
                 }
                 Some(doc) = change_stream.next() => {
                     info!("change detected: {:?}", doc.uri);
                     let path_buf = doc.uri.to_file_path().unwrap();
                     let path = path_buf.to_str().unwrap();
                     let contents = Some(doc.text);
                     update_input(workspace.clone(), db, path, contents).await;
+                    let _ = tx_needs_diagnostics.send(path.to_string());
+                }
+                Some(files_need_diagnostics) = diagnostics_stream.next() => {
+                    // let files_need_diagnostics = files_need_diagnostics.await;
+                    info!("files need diagnostics: {:?}", files_need_diagnostics);
+                    let mut ingots_need_diagnostics = FxHashSet::default();
+                    for file in files_need_diagnostics {
+                        let workspace = workspace.clone();
+                        let workspace = workspace.read().await;
+                        let ingot = workspace.get_ingot_for_file_path(&file).unwrap();
+                        ingots_need_diagnostics.insert(ingot);
+                    }
 
-                    let db = db.snapshot();
-                    let client = client.clone();
-                    let workspace = workspace.clone();
-                    self.workers.spawn(
-                        async move { handle_diagnostics(client, workspace, db, doc.uri).await }
-                    );
+                    info!("ingots need diagnostics: {:?}", ingots_need_diagnostics);
+                    for ingot in ingots_need_diagnostics.into_iter() {
+                        for file in ingot.files(db.as_input_db()) {
+                            let file = file.clone();
+                            let path = file.path(db.as_input_db());
+                            let path = lsp_types::Url::from_file_path(path).unwrap();
+                            let db = db.snapshot();
+                            let client = client.clone();
+                            let workspace = workspace.clone();
+                            let _ = self.workers.spawn(
+                                async move { handle_diagnostics(client.clone(), workspace.clone(), db, path).await }
+                            );
+                        }
+                    }
                 }
                 Some((params, responder)) = hover_stream.next() => {
                     let db = db.snapshot();
                     let workspace = workspace.clone();
-                    // let response = handle_hover(db, workspace, params).await;
                     let response = match self.workers.spawn(handle_hover(db, workspace, params)).await {
                         Ok(response) => response,
                         Err(e) => {
@@ -236,7 +266,7 @@ async fn update_input(
 ) {
     info!("updating input for {:?}", path);
     let workspace = &mut workspace.write().await;
-    let input = workspace.touch_input_from_file_path(db, path).unwrap();
+    let input = workspace.touch_input_for_file_path(db, path).unwrap();
     if let Some(contents) = contents {
         let _ = input.sync_from_text(db, contents);
     }

crates/language-server/src/goto.rs

@@ -132,13 +132,13 @@ mod tests {
         let ingot_base_dir = Path::new(&cargo_manifest_dir).join("test_files/single_ingot");
 
         let db = &mut LanguageServerDatabase::default();
-        let workspace = &mut Workspace::default(db);
+        let workspace = &mut Workspace::default();
 
-        let _ = workspace.set_workspace_root(db, ingot_base_dir.clone());
+        let _ = workspace.set_workspace_root(db, &ingot_base_dir);
 
         let fe_source_path = ingot_base_dir.join(fixture.path());
         let fe_source_path = fe_source_path.to_str().unwrap();
-        let input = workspace.touch_input_from_file_path(db, fixture.path());
+        let input = workspace.touch_input_for_file_path(db, fixture.path());
         assert_eq!(input.unwrap().ingot(db).kind(db), IngotKind::Local);
 
         input
@@ -149,7 +149,7 @@ mod tests {
             .top_mod_from_file_path(db.as_lower_hir_db(), fe_source_path)
             .unwrap();
 
-        let ingot = workspace.touch_ingot_from_file_path(db, fixture.path());
+        let ingot = workspace.touch_ingot_for_file_path(db, fixture.path());
         assert_eq!(ingot.unwrap().kind(db), IngotKind::Local);
 
         let cursors = extract_multiple_cursor_positions_from_spans(db, top_mod);
@@ -198,9 +198,9 @@ mod tests {
     )]
     fn test_goto_enclosing_path(fixture: Fixture<&str>) {
         let db = &mut LanguageServerDatabase::default();
-        let workspace = &mut Workspace::default(db);
+        let workspace = &mut Workspace::default();
         let input = workspace
-            .touch_input_from_file_path(db, fixture.path())
+            .touch_input_for_file_path(db, fixture.path())
             .unwrap();
         input.set_text(db).to((*fixture.content()).to_string());
         let top_mod = workspace
@@ -253,10 +253,10 @@ mod tests {
     )]
     fn test_smallest_enclosing_path(fixture: Fixture<&str>) {
         let db = &mut LanguageServerDatabase::default();
-        let workspace = &mut Workspace::default(db);
+        let workspace = &mut Workspace::default();
 
         workspace
-            .touch_input_from_file_path(db, fixture.path())
+            .touch_input_for_file_path(db, fixture.path())
             .unwrap()
             .set_text(db)
             .to((*fixture.content()).to_string());

crates/language-server/src/handlers/request.rs

@@ -34,7 +34,7 @@ pub async fn handle_hover(
         .uri
         .path();
     info!("getting hover info for file_path: {:?}", file_path);
-    let input = workspace.get_input_from_file_path(file_path);
+    let input = workspace.get_input_for_file_path(file_path);
     let ingot = input.map(|input| input.ingot(db.as_input_db()));
 
     let file_text = input.unwrap().text(db.as_input_db());