Analyzing progress notifications

src/config.rs

@@ -97,6 +97,9 @@ impl Config {
                     response.pop_front().as_ref().and_then(Value::as_bool).unwrap_or_default();
                 state.config.enable_proc_macros =
                     response.pop_front().as_ref().and_then(Value::as_bool).unwrap_or(false);
+                state
+                    .analysis_progress_controller
+                    .set_procmacros_enabled(state.config.enable_proc_macros);
 
                 debug!("reloaded configuration: {:#?}", state.config);
 

src/ide/analysis_progress.rs

@@ -0,0 +1,154 @@
+use std::collections::HashSet;
+use std::sync::{Arc, Mutex};
+
+use lsp_types::notification::Notification;
+
+use crate::id_generator::IdGenerator;
+use crate::server::client::Notifier;
+use crate::state::Beacon;
+
+/// Controller used to send notifications to the client about analysis progress.
+/// Uses information provided from other controllers (diagnostics controller, procmacro controller)
+/// to assess if diagnostics are in fact calculated.
+#[derive(Debug, Clone)]
+pub struct AnalysisProgressController {
+    notifier: Notifier,
+    /// ID of the diagnostics "generation" - the scheduled diagnostics jobs set.
+    /// Used to filter out stale threads finishing when new ones (from newer "generation")
+    /// are already in progress and being tracked by the controller.
+    generation_id: Arc<Mutex<u64>>,
+    /// Sequential IDs of state snapshots from the current generation, used to track their status
+    /// (present meaning it's still being used)
+    active_snapshots: Arc<Mutex<HashSet<usize>>>,
+    id_generator: Arc<IdGenerator>,
+    /// If `true` - a request to procmacro server was submitted, meaning that analysis will extend
+    /// beyond the current generation of diagnostics.
+    did_submit_procmacro_request: Arc<Mutex<bool>>,
+    /// Indicates that a notification was sent and analysis (i.e. macro expansion) is taking place.
+    analysis_in_progress: Arc<Mutex<bool>>,
+    /// Loaded asynchronously from config - unset if config was not loaded yet.
+    /// Has to be set in order for analysis to finish.
+    procmacros_enabled: Arc<Mutex<Option<bool>>>,
+}
+
+impl AnalysisProgressController {
+    pub fn new(notifier: Notifier) -> Self {
+        let id_generator = Arc::new(IdGenerator::default());
+        Self {
+            notifier,
+            id_generator: id_generator.clone(),
+            active_snapshots: Arc::new(Mutex::new(HashSet::default())),
+            did_submit_procmacro_request: Arc::new(Mutex::new(true)),
+            analysis_in_progress: Arc::new(Mutex::new(false)),
+            procmacros_enabled: Arc::new(Mutex::new(None)),
+            generation_id: Arc::new(Mutex::new(id_generator.unique_id())),
+        }
+    }
+
+    /// Signals that a request to proc macro server was made during the current generation of
+    /// diagnostics.
+    pub fn register_procmacro_request(&self) {
+        let mut write_guard = self.did_submit_procmacro_request.lock().unwrap();
+        *write_guard = true;
+    }
+
+    /// Allows to set the procmacro configuration to whatever is in the config, upon loading it.
+    pub fn set_procmacros_enabled(&self, value: bool) {
+        let mut guard = self.procmacros_enabled.lock().unwrap();
+        *guard = Some(value);
+    }
+
+    /// Sets handlers for tracking beacons sent to threads.
+    /// The beacons are wrapping snapshots, which are signalling when diagnostics finished
+    /// calculating for a given snapshot (used for calculating files diagnostics or removing
+    /// stale ones)
+    pub fn track_analysis<T: Send>(&self, beacons: &mut [Beacon<T>]) {
+        let gen_id = self.next_generation_id();
+
+        self.clear_active_snapshots();
+        beacons.iter_mut().enumerate().for_each(|(i, beacon)| {
+            self.insert_active_snapshot(i);
+
+            let self_ref: AnalysisProgressController = self.clone();
+            beacon.on_signal(move || {
+                let current_gen = self_ref.get_generation_id();
+                if current_gen == gen_id {
+                    self_ref.remove_active_snapshot(i);
+                    self_ref.try_stop_analysis();
+                }
+            });
+        });
+
+        self.start_analysis();
+    }
+
+    fn insert_active_snapshot(&self, snapshot_id: usize) {
+        let mut active_snapshots = self.active_snapshots.lock().unwrap();
+        active_snapshots.insert(snapshot_id);
+    }
+
+    fn next_generation_id(&self) -> u64 {
+        let mut generation_id_guard = self.generation_id.lock().unwrap();
+        *generation_id_guard = self.id_generator.unique_id();
+        *generation_id_guard
+    }
+
+    fn get_generation_id(&self) -> u64 {
+        *self.generation_id.lock().unwrap()
+    }
+
+    fn remove_active_snapshot(&self, snapshot_id: usize) {
+        let mut active_snapshots = self.active_snapshots.lock().unwrap();
+        active_snapshots.remove(&snapshot_id);
+    }
+
+    fn clear_active_snapshots(&self) {
+        let active_snapshots_ref = self.active_snapshots.clone();
+        active_snapshots_ref.lock().unwrap().clear();
+    }
+
+    /// Starts a next generation of diagnostics, sends a notification
+    fn start_analysis(&self) {
+        let mut analysis_in_progress = self.analysis_in_progress.lock().unwrap();
+        if !(*analysis_in_progress) {
+            *analysis_in_progress = true;
+            self.notifier.notify::<DiagnosticsCalculationStart>(());
+        }
+    }
+
+    /// Checks a bunch of conditions and if they are fulfilled, sends stop notification
+    /// and resets the state back to start of generation defaults.
+    fn try_stop_analysis(&self) {
+        let mut did_submit_procmacro_request = self.did_submit_procmacro_request.lock().unwrap();
+        let snapshots_empty = self.active_snapshots.lock().unwrap().is_empty();
+        let mut analysis_in_progress = self.analysis_in_progress.lock().unwrap();
+        let procmacros_enabled = *self.procmacros_enabled.lock().unwrap();
+
+        if snapshots_empty
+            && (!*did_submit_procmacro_request || (procmacros_enabled == Some(false)))
+            && *analysis_in_progress
+        {
+            *analysis_in_progress = false;
+            *did_submit_procmacro_request = false;
+            self.notifier.notify::<DiagnosticsCalculationFinish>(());
+        }
+    }
+}
+
+/// Notifies about diagnostics generation which is beginning to calculate
+#[derive(Debug)]
+pub struct DiagnosticsCalculationStart;
+
+impl Notification for DiagnosticsCalculationStart {
+    type Params = ();
+    const METHOD: &'static str = "cairo/diagnosticsCalculationStart";
+}
+
+/// Notifies about diagnostics generation which ended calculating
+#[derive(Debug)]
+pub struct DiagnosticsCalculationFinish;
+
+impl Notification for DiagnosticsCalculationFinish {
+    type Params = ();
+    const METHOD: &'static str = "cairo/diagnosticsCalculationFinish";
+}

src/ide/mod.rs

@@ -1,3 +1,4 @@
+pub mod analysis_progress;
 pub mod code_actions;
 pub mod completion;
 pub mod formatter;

src/lang/diagnostics/mod.rs

@@ -11,6 +11,7 @@ use tracing::{error, trace};
 use self::project_diagnostics::ProjectDiagnostics;
 use self::refresh::{clear_old_diagnostics, refresh_diagnostics};
 use self::trigger::trigger;
+use crate::ide::analysis_progress::AnalysisProgressController;
 use crate::lang::diagnostics::file_batches::{batches, find_primary_files, find_secondary_files};
 use crate::lang::lsp::LsProtoGroup;
 use crate::server::client::Notifier;
@@ -41,9 +42,17 @@ pub struct DiagnosticsController {
 
 impl DiagnosticsController {
     /// Creates a new diagnostics controller.
-    pub fn new(notifier: Notifier) -> Self {
+    pub fn new(
+        notifier: Notifier,
+        analysis_progress_controller: AnalysisProgressController,
+    ) -> Self {
         let (trigger, receiver) = trigger();
-        let (thread, parallelism) = DiagnosticsControllerThread::spawn(receiver, notifier);
+        let (thread, parallelism) = DiagnosticsControllerThread::spawn(
+            receiver,
+            notifier,
+            analysis_progress_controller,
+        );
+
         Self {
             trigger,
             _thread: thread,
@@ -53,8 +62,7 @@ impl DiagnosticsController {
 
     /// Schedules diagnostics refreshing on snapshot(s) of the current state.
     pub fn refresh(&self, state: &State) {
-        let state_snapshots = StateSnapshots::new(state, &self.state_snapshots_props);
-        self.trigger.activate(state_snapshots);
+        self.trigger.activate(StateSnapshots::new(state, &self.state_snapshots_props));
     }
 }
 
@@ -64,6 +72,7 @@ struct DiagnosticsControllerThread {
     notifier: Notifier,
     pool: thread::Pool,
     project_diagnostics: ProjectDiagnostics,
+    analysis_progress_controller: AnalysisProgressController,
 }
 
 impl DiagnosticsControllerThread {
@@ -72,10 +81,12 @@ impl DiagnosticsControllerThread {
     fn spawn(
         receiver: trigger::Receiver<StateSnapshots>,
         notifier: Notifier,
+        analysis_progress_controller: AnalysisProgressController,
     ) -> (JoinHandle, NonZero<usize>) {
         let this = Self {
             receiver,
             notifier,
+            analysis_progress_controller,
             pool: thread::Pool::new(),
             project_diagnostics: ProjectDiagnostics::new(),
         };
@@ -92,7 +103,8 @@ impl DiagnosticsControllerThread {
 
     /// Runs diagnostics controller's event loop.
     fn event_loop(&self) {
-        while let Some(state_snapshots) = self.receiver.wait() {
+        while let Some(mut state_snapshots) = self.receiver.wait() {
+            self.analysis_progress_controller.track_analysis(&mut state_snapshots.0);
             if let Err(err) = catch_unwind(AssertUnwindSafe(|| {
                 self.diagnostics_controller_tick(state_snapshots);
             })) {
@@ -109,7 +121,7 @@ impl DiagnosticsControllerThread {
     /// Runs a single tick of the diagnostics controller's event loop.
     #[tracing::instrument(skip_all)]
     fn diagnostics_controller_tick(&self, state_snapshots: StateSnapshots) {
-        let (state, primary_snapshots, secondary_snapshots) = state_snapshots.split();
+        let (mut state, primary_snapshots, secondary_snapshots) = state_snapshots.split();
 
         let primary_set = find_primary_files(&state.db, &state.open_files);
         let primary: Vec<_> = primary_set.iter().copied().collect();
@@ -126,6 +138,7 @@ impl DiagnosticsControllerThread {
 
         self.spawn_worker(move |project_diagnostics, notifier| {
             clear_old_diagnostics(files_to_preserve, project_diagnostics, notifier);
+            state.signal();
         });
     }
 
@@ -150,7 +163,7 @@ impl DiagnosticsControllerThread {
     fn spawn_refresh_worker(&self, files: &[FileId], state_snapshots: Vec<StateSnapshot>) {
         let files_batches = batches(files, self.pool.parallelism());
         assert_eq!(files_batches.len(), state_snapshots.len());
-        for (batch, state) in zip(files_batches, state_snapshots) {
+        for (batch, mut state) in zip(files_batches, state_snapshots) {
             self.spawn_worker(move |project_diagnostics, notifier| {
                 refresh_diagnostics(
                     &state.db,
@@ -159,6 +172,7 @@ impl DiagnosticsControllerThread {
                     project_diagnostics,
                     notifier,
                 );
+                state.signal();
             });
         }
     }

src/lang/proc_macros/client/mod.rs

@@ -16,8 +16,10 @@ use scarb_proc_macro_server_types::methods::expand::{
 pub use status::ClientStatus;
 use tracing::error;
 
+use crate::id_generator;
+use crate::ide::analysis_progress::AnalysisProgressController;
+
 pub mod connection;
-mod id_generator;
 pub mod status;
 
 #[derive(Debug)]
@@ -33,15 +35,21 @@ pub struct ProcMacroClient {
     id_generator: id_generator::IdGenerator,
     requests_params: Mutex<HashMap<RequestId, RequestParams>>,
     error_channel: Sender<()>,
+    analysis_progress_controller: AnalysisProgressController,
 }
 
 impl ProcMacroClient {
-    pub fn new(connection: ProcMacroServerConnection, error_channel: Sender<()>) -> Self {
+    pub fn new(
+        connection: ProcMacroServerConnection,
+        error_channel: Sender<()>,
+        analysis_progress_controller: AnalysisProgressController,
+    ) -> Self {
         Self {
             connection,
             id_generator: Default::default(),
             requests_params: Default::default(),
             error_channel,
+            analysis_progress_controller,
         }
     }
 
@@ -142,6 +150,7 @@ impl ProcMacroClient {
         match self.send_request_untracked::<M>(id, &params) {
             Ok(()) => {
                 requests_params.insert(id, map(params));
+                self.analysis_progress_controller.register_procmacro_request();
             }
             Err(err) => {
                 error!("Sending request to proc-macro-server failed: {err:?}");

src/lang/proc_macros/controller.rs

@@ -19,6 +19,7 @@ use super::client::connection::ProcMacroServerConnection;
 use super::client::status::ClientStatus;
 use super::client::{ProcMacroClient, RequestParams};
 use crate::config::Config;
+use crate::ide::analysis_progress::AnalysisProgressController;
 use crate::lang::db::AnalysisDatabase;
 use crate::lang::proc_macros::db::ProcMacroGroup;
 use crate::lang::proc_macros::plugins::proc_macro_plugin_suite;
@@ -53,17 +54,23 @@ pub struct ProcMacroClientController {
     plugin_suite: Option<PluginSuite>,
     initialization_retries: RateLimiter<NotKeyed, InMemoryState, QuantaClock>,
     channels: ProcMacroChannels,
+    analysis_progress_controller: AnalysisProgressController,
 }
 
 impl ProcMacroClientController {
     pub fn channels(&mut self) -> ProcMacroChannels {
         self.channels.clone()
     }
 
-    pub fn new(scarb: ScarbToolchain, notifier: Notifier) -> Self {
+    pub fn new(
+        scarb: ScarbToolchain,
+        notifier: Notifier,
+        analysis_progress_controller: AnalysisProgressController,
+    ) -> Self {
         Self {
             scarb,
             notifier,
+            analysis_progress_controller,
             plugin_suite: Default::default(),
             initialization_retries: RateLimiter::direct(
                 Quota::with_period(Duration::from_secs(
@@ -170,10 +177,10 @@ impl ProcMacroClientController {
                         self.channels.response_sender.clone(),
                     ),
                     self.channels.error_sender.clone(),
+                    self.analysis_progress_controller.clone(),
                 );
 
                 client.start_initialize();
-
                 db.set_proc_macro_client_status(ClientStatus::Starting(Arc::new(client)));
             }
             Err(err) => {

src/lib.rs

@@ -71,6 +71,7 @@ use crate::state::State;
 
 mod config;
 mod env_config;
+mod id_generator;
 mod ide;
 mod lang;
 pub mod lsp;

src/server/client.rs

@@ -26,7 +26,7 @@ pub struct Client<'s> {
     pub(super) requester: Requester<'s>,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct Notifier(ClientSender);
 
 #[derive(Clone)]