From 70730d7655bf2178d88938651635843b7ef015c8 Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Tue, 31 Mar 2020 17:05:15 +0200 Subject: Rename cargo_watch -> flycheck --- crates/ra_flycheck/src/lib.rs | 374 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 374 insertions(+) create mode 100644 crates/ra_flycheck/src/lib.rs (limited to 'crates/ra_flycheck/src/lib.rs') diff --git a/crates/ra_flycheck/src/lib.rs b/crates/ra_flycheck/src/lib.rs new file mode 100644 index 000000000..77ede8f63 --- /dev/null +++ b/crates/ra_flycheck/src/lib.rs @@ -0,0 +1,374 @@ +//! cargo_check provides the functionality needed to run `cargo check` or +//! another compatible command (f.x. clippy) in a background thread and provide +//! LSP diagnostics based on the output of the command. +use cargo_metadata::Message; +use crossbeam_channel::{never, select, unbounded, Receiver, RecvError, Sender}; +use lsp_types::{ + CodeAction, CodeActionOrCommand, Diagnostic, Url, WorkDoneProgress, WorkDoneProgressBegin, + WorkDoneProgressEnd, WorkDoneProgressReport, +}; +use std::{ + error, fmt, + io::{BufRead, BufReader}, + path::{Path, PathBuf}, + process::{Command, Stdio}, + time::Instant, +}; + +mod conv; + +use crate::conv::{map_rust_diagnostic_to_lsp, MappedRustDiagnostic}; + +pub use crate::conv::url_from_path_with_drive_lowercasing; + +#[derive(Clone, Debug)] +pub struct CheckConfig { + pub enable: bool, + pub args: Vec, + pub command: String, + pub all_targets: bool, +} + +/// CheckWatcher wraps the shared state and communication machinery used for +/// running `cargo check` (or other compatible command) and providing +/// diagnostics based on the output. +/// The spawned thread is shut down when this struct is dropped. +#[derive(Debug)] +pub struct CheckWatcher { + // XXX: drop order is significant + cmd_send: Sender, + handle: Option>, + pub task_recv: Receiver, +} + +impl CheckWatcher { + pub fn new(config: CheckConfig, workspace_root: PathBuf) -> CheckWatcher { + let (task_send, task_recv) = unbounded::(); + let (cmd_send, cmd_recv) = unbounded::(); + let handle = jod_thread::spawn(move || { + let mut check = CheckWatcherThread::new(config, workspace_root); + check.run(&task_send, &cmd_recv); + }); + CheckWatcher { task_recv, cmd_send, handle: Some(handle) } + } + + /// Schedule a re-start of the cargo check worker. + pub fn update(&self) { + self.cmd_send.send(CheckCommand::Update).unwrap(); + } +} + +#[derive(Debug)] +pub enum CheckTask { + /// Request a clearing of all cached diagnostics from the check watcher + ClearDiagnostics, + + /// Request adding a diagnostic with fixes included to a file + AddDiagnostic { url: Url, diagnostic: Diagnostic, fixes: Vec }, + + /// Request check progress notification to client + Status(WorkDoneProgress), +} + +pub enum CheckCommand { + /// Request re-start of check thread + Update, +} + +struct CheckWatcherThread { + options: CheckConfig, + workspace_root: PathBuf, + watcher: WatchThread, + last_update_req: Option, +} + +impl CheckWatcherThread { + fn new(options: CheckConfig, workspace_root: PathBuf) -> CheckWatcherThread { + CheckWatcherThread { + options, + workspace_root, + watcher: WatchThread::dummy(), + last_update_req: None, + } + } + + fn run(&mut self, task_send: &Sender, cmd_recv: &Receiver) { + // If we rerun the thread, we need to discard the previous check results first + self.clean_previous_results(task_send); + + loop { + select! { + recv(&cmd_recv) -> cmd => match cmd { + Ok(cmd) => self.handle_command(cmd), + Err(RecvError) => { + // Command channel has closed, so shut down + break; + }, + }, + recv(self.watcher.message_recv) -> msg => match msg { + Ok(msg) => self.handle_message(msg, task_send), + Err(RecvError) => { + // Watcher finished, replace it with a never channel to + // avoid busy-waiting. + std::mem::replace(&mut self.watcher.message_recv, never()); + }, + } + }; + + if self.should_recheck() { + self.last_update_req.take(); + task_send.send(CheckTask::ClearDiagnostics).unwrap(); + + // Replace with a dummy watcher first so we drop the original and wait for completion + std::mem::replace(&mut self.watcher, WatchThread::dummy()); + + // Then create the actual new watcher + self.watcher = WatchThread::new(&self.options, &self.workspace_root); + } + } + } + + fn clean_previous_results(&self, task_send: &Sender) { + task_send.send(CheckTask::ClearDiagnostics).unwrap(); + task_send + .send(CheckTask::Status(WorkDoneProgress::End(WorkDoneProgressEnd { message: None }))) + .unwrap(); + } + + fn should_recheck(&mut self) -> bool { + if let Some(_last_update_req) = &self.last_update_req { + // We currently only request an update on save, as we need up to + // date source on disk for cargo check to do it's magic, so we + // don't really need to debounce the requests at this point. + return true; + } + false + } + + fn handle_command(&mut self, cmd: CheckCommand) { + match cmd { + CheckCommand::Update => self.last_update_req = Some(Instant::now()), + } + } + + fn handle_message(&self, msg: CheckEvent, task_send: &Sender) { + match msg { + CheckEvent::Begin => { + task_send + .send(CheckTask::Status(WorkDoneProgress::Begin(WorkDoneProgressBegin { + title: "Running 'cargo check'".to_string(), + cancellable: Some(false), + message: None, + percentage: None, + }))) + .unwrap(); + } + + CheckEvent::End => { + task_send + .send(CheckTask::Status(WorkDoneProgress::End(WorkDoneProgressEnd { + message: None, + }))) + .unwrap(); + } + + CheckEvent::Msg(Message::CompilerArtifact(msg)) => { + task_send + .send(CheckTask::Status(WorkDoneProgress::Report(WorkDoneProgressReport { + cancellable: Some(false), + message: Some(msg.target.name), + percentage: None, + }))) + .unwrap(); + } + + CheckEvent::Msg(Message::CompilerMessage(msg)) => { + let map_result = map_rust_diagnostic_to_lsp(&msg.message, &self.workspace_root); + if map_result.is_empty() { + return; + } + + for MappedRustDiagnostic { location, diagnostic, fixes } in map_result { + let fixes = fixes + .into_iter() + .map(|fix| { + CodeAction { diagnostics: Some(vec![diagnostic.clone()]), ..fix }.into() + }) + .collect(); + + task_send + .send(CheckTask::AddDiagnostic { url: location.uri, diagnostic, fixes }) + .unwrap(); + } + } + + CheckEvent::Msg(Message::BuildScriptExecuted(_msg)) => {} + CheckEvent::Msg(Message::Unknown) => {} + } + } +} + +#[derive(Debug)] +pub struct DiagnosticWithFixes { + diagnostic: Diagnostic, + fixes: Vec, +} + +/// WatchThread exists to wrap around the communication needed to be able to +/// run `cargo check` without blocking. Currently the Rust standard library +/// doesn't provide a way to read sub-process output without blocking, so we +/// have to wrap sub-processes output handling in a thread and pass messages +/// back over a channel. +/// The correct way to dispose of the thread is to drop it, on which the +/// sub-process will be killed, and the thread will be joined. +struct WatchThread { + // XXX: drop order is significant + message_recv: Receiver, + _handle: Option>, +} + +enum CheckEvent { + Begin, + Msg(cargo_metadata::Message), + End, +} + +#[derive(Debug)] +pub struct CargoError(String); + +impl fmt::Display for CargoError { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Cargo failed: {}", self.0) + } +} +impl error::Error for CargoError {} + +pub fn run_cargo( + args: &[String], + current_dir: Option<&Path>, + on_message: &mut dyn FnMut(cargo_metadata::Message) -> bool, +) -> Result<(), CargoError> { + let mut command = Command::new("cargo"); + if let Some(current_dir) = current_dir { + command.current_dir(current_dir); + } + + let mut child = command + .args(args) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .stdin(Stdio::null()) + .spawn() + .expect("couldn't launch cargo"); + + // We manually read a line at a time, instead of using serde's + // stream deserializers, because the deserializer cannot recover + // from an error, resulting in it getting stuck, because we try to + // be resillient against failures. + // + // Because cargo only outputs one JSON object per line, we can + // simply skip a line if it doesn't parse, which just ignores any + // erroneus output. + let stdout = BufReader::new(child.stdout.take().unwrap()); + let mut read_at_least_one_message = false; + + for line in stdout.lines() { + let line = match line { + Ok(line) => line, + Err(err) => { + log::error!("Couldn't read line from cargo: {}", err); + continue; + } + }; + + let message = serde_json::from_str::(&line); + let message = match message { + Ok(message) => message, + Err(err) => { + log::error!("Invalid json from cargo check, ignoring ({}): {:?} ", err, line); + continue; + } + }; + + read_at_least_one_message = true; + + if !on_message(message) { + break; + } + } + + // It is okay to ignore the result, as it only errors if the process is already dead + let _ = child.kill(); + + let err_msg = match child.wait() { + Ok(exit_code) if !exit_code.success() && !read_at_least_one_message => { + // FIXME: Read the stderr to display the reason, see `read2()` reference in PR comment: + // https://github.com/rust-analyzer/rust-analyzer/pull/3632#discussion_r395605298 + format!( + "the command produced no valid metadata (exit code: {:?}): cargo {}", + exit_code, + args.join(" ") + ) + } + Err(err) => format!("io error: {:?}", err), + Ok(_) => return Ok(()), + }; + + Err(CargoError(err_msg)) +} + +impl WatchThread { + fn dummy() -> WatchThread { + WatchThread { message_recv: never(), _handle: None } + } + + fn new(options: &CheckConfig, workspace_root: &Path) -> WatchThread { + let mut args: Vec = vec![ + options.command.clone(), + "--workspace".to_string(), + "--message-format=json".to_string(), + "--manifest-path".to_string(), + format!("{}/Cargo.toml", workspace_root.display()), + ]; + if options.all_targets { + args.push("--all-targets".to_string()); + } + args.extend(options.args.iter().cloned()); + + let (message_send, message_recv) = unbounded(); + let workspace_root = workspace_root.to_owned(); + let handle = if options.enable { + Some(jod_thread::spawn(move || { + // If we trigger an error here, we will do so in the loop instead, + // which will break out of the loop, and continue the shutdown + let _ = message_send.send(CheckEvent::Begin); + + let res = run_cargo(&args, Some(&workspace_root), &mut |message| { + // Skip certain kinds of messages to only spend time on what's useful + match &message { + Message::CompilerArtifact(artifact) if artifact.fresh => return true, + Message::BuildScriptExecuted(_) => return true, + Message::Unknown => return true, + _ => {} + } + + // if the send channel was closed, we want to shutdown + message_send.send(CheckEvent::Msg(message)).is_ok() + }); + + if let Err(err) = res { + // FIXME: make the `message_send` to be `Sender>` + // to display user-caused misconfiguration errors instead of just logging them here + log::error!("Cargo watcher failed {:?}", err); + } + + // We can ignore any error here, as we are already in the progress + // of shutting down. + let _ = message_send.send(CheckEvent::End); + })) + } else { + None + }; + WatchThread { message_recv, _handle: handle } + } +} -- cgit v1.2.3