batched ingot-wide diagnostics; cleanup

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 common::InputDb;
 use fork_stream::StreamExt as _;
+use futures_batch::ChunksTimeoutStreamExt;
+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;
@@ -23,6 +26,8 @@ use crate::workspace::{IngotFileContext, SyncableInputFile, Workspace};
 
 use log::info;
 
+// use tokio_stream::StreamExt;
+
 use tower_lsp::Client;
 
 pub struct Backend {
@@ -58,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
@@ -83,11 +85,13 @@ impl Backend {
                 Box::pin(async move { matches!(change_type, lsp_types::FileChangeType::CREATED) })
             });
 
-        let mut did_delete_watch_file_stream =
-            flat_did_change_watched_files.clone().filter(|change| {
+        let mut did_delete_watch_file_stream = 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();
@@ -128,7 +132,12 @@ 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(500, std::time::Duration::from_millis(30))
             .fuse();
 
         let mut hover_stream = messaging.hover_stream.fuse();
@@ -140,8 +149,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
@@ -180,25 +187,44 @@ 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() => {
+                    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;
+                            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();
+                            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) => {

crates/language-server/src/diagnostics.rs

@@ -135,7 +135,7 @@ fn run_diagnostics(
 ) -> Vec<common::diagnostics::CompleteDiagnostic> {
     let file_path = path;
     let top_mod = workspace
-        .top_mod_for_file_path(db.as_lower_hir_db(), file_path)
+        .top_mod_from_file_path(db.as_lower_hir_db(), file_path)
         .unwrap();
     let diags = &db.analyze_top_mod(top_mod);
     db.finalize_diags(diags)

crates/language-server/src/goto.rs

@@ -132,9 +132,9 @@ 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();
@@ -146,7 +146,7 @@ mod tests {
             .set_text(db)
             .to((*fixture.content()).to_string());
         let top_mod = workspace
-            .top_mod_for_file_path(db.as_lower_hir_db(), fe_source_path)
+            .top_mod_from_file_path(db.as_lower_hir_db(), fe_source_path)
             .unwrap();
 
         let ingot = workspace.touch_ingot_for_file_path(db, fixture.path());
@@ -198,13 +198,13 @@ 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_for_file_path(db, fixture.path())
             .unwrap();
         input.set_text(db).to((*fixture.content()).to_string());
         let top_mod = workspace
-            .top_mod_for_file_path(db.as_lower_hir_db(), fixture.path())
+            .top_mod_from_file_path(db.as_lower_hir_db(), fixture.path())
             .unwrap();
 
         let cursors = extract_multiple_cursor_positions_from_spans(db, top_mod);
@@ -253,15 +253,15 @@ 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_for_file_path(db, fixture.path())
             .unwrap()
             .set_text(db)
             .to((*fixture.content()).to_string());
         let top_mod = workspace
-            .top_mod_for_file_path(db.as_lower_hir_db(), fixture.path())
+            .top_mod_from_file_path(db.as_lower_hir_db(), fixture.path())
             .unwrap();
 
         let cursors = extract_multiple_cursor_positions_from_spans(db, top_mod);

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

@@ -18,6 +18,7 @@ use crate::{
     util::{to_lsp_location_from_scope, to_offset_from_position},
     workspace::{IngotFileContext, Workspace},
 };
+// use tower_lsp::lsp_types::{ResponseError, Url};
 
 pub async fn handle_hover(
     db: Snapshot<LanguageServerDatabase>,
@@ -42,10 +43,14 @@ pub async fn handle_hover(
         .nth(params.text_document_position_params.position.line as usize)
         .unwrap();
 
+    // let cursor: Cursor = params.text_document_position_params.position.into();
     let cursor: Cursor = to_offset_from_position(
         params.text_document_position_params.position,
         file_text.as_str(),
     );
+    // let file_path = std::path::Path::new(file_path);
+
+    // info!("got ingot: {:?} of type {:?}", ingot, ingot.map(|ingot| ingot.kind(&mut state.db)));
 
     let ingot_info: Option<String> = {
         let ingot_type = match ingot {
@@ -70,7 +75,7 @@ pub async fn handle_hover(
     };
 
     let top_mod = workspace
-        .top_mod_for_file_path(db.as_lower_hir_db(), file_path)
+        .top_mod_from_file_path(db.as_lower_hir_db(), file_path)
         .unwrap();
     let early_resolution = goto_enclosing_path(&db, top_mod, cursor);
 
@@ -119,7 +124,7 @@ pub async fn handle_goto_definition(
     // Get the module and the goto info
     let file_path = params.text_document.uri.path();
     let top_mod = workspace
-        .top_mod_for_file_path(db.as_lower_hir_db(), file_path)
+        .top_mod_from_file_path(db.as_lower_hir_db(), file_path)
         .unwrap();
     let goto_info = goto_enclosing_path(&db, top_mod, cursor);
 

crates/language-server/src/main.rs

@@ -6,6 +6,7 @@ mod globals;
 mod goto;
 mod language_server;
 mod logger;
+mod stream_buffer_until;
 mod util;
 mod workspace;