From 80ef52f0d5f9383e9f42ae77d9cb6b77483f11a6 Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Sun, 28 Jun 2020 20:00:04 +0200 Subject: [PATCH 1/5] Make sure to join the child --- crates/flycheck/src/lib.rs | 27 +++++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/crates/flycheck/src/lib.rs b/crates/flycheck/src/lib.rs index 92ec4f92e3..9335098ff0 100644 --- a/crates/flycheck/src/lib.rs +++ b/crates/flycheck/src/lib.rs @@ -5,8 +5,9 @@ use std::{ fmt, io::{self, BufReader}, + ops, path::PathBuf, - process::{Command, Stdio}, + process::{self, Command, Stdio}, time::Duration, }; @@ -236,8 +237,9 @@ fn run_cargo( mut command: Command, on_message: &mut dyn FnMut(cargo_metadata::Message) -> bool, ) -> io::Result<()> { - let mut child = + let child = command.stdout(Stdio::piped()).stderr(Stdio::null()).stdin(Stdio::null()).spawn()?; + let mut child = ChildKiller(child); // We manually read a line at a time, instead of using serde's // stream deserializers, because the deserializer cannot recover @@ -283,3 +285,24 @@ fn run_cargo( Ok(()) } + +struct ChildKiller(process::Child); + +impl ops::Deref for ChildKiller { + type Target = process::Child; + fn deref(&self) -> &process::Child { + &self.0 + } +} + +impl ops::DerefMut for ChildKiller { + fn deref_mut(&mut self) -> &mut process::Child { + &mut self.0 + } +} + +impl Drop for ChildKiller { + fn drop(&mut self) { + let _ = self.0.kill(); + } +} From 309b21f37861a8c6550f93e7f9b8f955a0b4b256 Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Sun, 28 Jun 2020 22:31:40 +0200 Subject: [PATCH 2/5] Rename --- crates/flycheck/src/lib.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/flycheck/src/lib.rs b/crates/flycheck/src/lib.rs index 9335098ff0..0d68fcd4d3 100644 --- a/crates/flycheck/src/lib.rs +++ b/crates/flycheck/src/lib.rs @@ -51,7 +51,7 @@ impl fmt::Display for FlycheckConfig { pub struct FlycheckHandle { // XXX: drop order is significant cmd_send: Sender, - handle: jod_thread::JoinHandle, + thread: jod_thread::JoinHandle, } impl FlycheckHandle { @@ -61,10 +61,10 @@ impl FlycheckHandle { workspace_root: PathBuf, ) -> FlycheckHandle { let (cmd_send, cmd_recv) = unbounded::(); - let handle = jod_thread::spawn(move || { + let thread = jod_thread::spawn(move || { FlycheckActor::new(sender, config, workspace_root).run(cmd_recv); }); - FlycheckHandle { cmd_send, handle } + FlycheckHandle { cmd_send, thread } } /// Schedule a re-start of the cargo check worker. From eddb744d9038306d020ac46d12a373508e9d3268 Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Sun, 28 Jun 2020 22:35:18 +0200 Subject: [PATCH 3/5] Naming --- crates/flycheck/src/lib.rs | 15 +++++++-------- crates/vfs-notify/src/lib.rs | 8 ++++---- 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/crates/flycheck/src/lib.rs b/crates/flycheck/src/lib.rs index 0d68fcd4d3..073bcc9aec 100644 --- a/crates/flycheck/src/lib.rs +++ b/crates/flycheck/src/lib.rs @@ -50,7 +50,7 @@ impl fmt::Display for FlycheckConfig { #[derive(Debug)] pub struct FlycheckHandle { // XXX: drop order is significant - cmd_send: Sender, + sender: Sender, thread: jod_thread::JoinHandle, } @@ -60,16 +60,15 @@ impl FlycheckHandle { config: FlycheckConfig, workspace_root: PathBuf, ) -> FlycheckHandle { - let (cmd_send, cmd_recv) = unbounded::(); - let thread = jod_thread::spawn(move || { - FlycheckActor::new(sender, config, workspace_root).run(cmd_recv); - }); - FlycheckHandle { cmd_send, thread } + let actor = FlycheckActor::new(sender, config, workspace_root); + let (sender, receiver) = unbounded::(); + let thread = jod_thread::spawn(move || actor.run(receiver)); + FlycheckHandle { sender, thread } } /// Schedule a re-start of the cargo check worker. pub fn update(&self) { - self.cmd_send.send(Restart).unwrap(); + self.sender.send(Restart).unwrap(); } } @@ -125,7 +124,7 @@ impl FlycheckActor { recv(check_chan.unwrap_or(&never())) -> msg => Some(Event::CheckEvent(msg.ok())), } } - fn run(&mut self, inbox: Receiver) { + fn run(mut self, inbox: Receiver) { while let Some(event) = self.next_event(&inbox) { match event { Event::Restart(Restart) => { diff --git a/crates/vfs-notify/src/lib.rs b/crates/vfs-notify/src/lib.rs index 25ba8d7982..b1ea298ae5 100644 --- a/crates/vfs-notify/src/lib.rs +++ b/crates/vfs-notify/src/lib.rs @@ -10,7 +10,7 @@ mod include; use std::convert::{TryFrom, TryInto}; -use crossbeam_channel::{select, unbounded, Receiver}; +use crossbeam_channel::{select, unbounded, Receiver, Sender}; use notify::{RecommendedWatcher, RecursiveMode, Watcher}; use paths::{AbsPath, AbsPathBuf}; use rustc_hash::FxHashSet; @@ -22,8 +22,8 @@ use crate::include::Include; #[derive(Debug)] pub struct NotifyHandle { // Relative order of fields below is significant. - sender: crossbeam_channel::Sender, - _thread: jod_thread::JoinHandle, + sender: Sender, + thread: jod_thread::JoinHandle, } #[derive(Debug)] @@ -37,7 +37,7 @@ impl loader::Handle for NotifyHandle { let actor = NotifyActor::new(sender); let (sender, receiver) = unbounded::(); let thread = jod_thread::spawn(move || actor.run(receiver)); - NotifyHandle { sender, _thread: thread } + NotifyHandle { sender, thread } } fn set_config(&mut self, config: loader::Config) { self.sender.send(Message::Config(config)).unwrap() From 32e85a1a89877dc1314ea950bd4cba43d9ad9627 Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Sun, 28 Jun 2020 23:01:28 +0200 Subject: [PATCH 4/5] More standard pattern for Cargo --- crates/flycheck/src/lib.rs | 161 +++++++++++++++++++------------------ 1 file changed, 84 insertions(+), 77 deletions(-) diff --git a/crates/flycheck/src/lib.rs b/crates/flycheck/src/lib.rs index 073bcc9aec..ab1d71b981 100644 --- a/crates/flycheck/src/lib.rs +++ b/crates/flycheck/src/lib.rs @@ -100,8 +100,7 @@ struct FlycheckActor { /// 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. - // XXX: drop order is significant - check_process: Option<(Receiver, jod_thread::JoinHandle)>, + check_process: Option, } enum Event { @@ -118,7 +117,7 @@ impl FlycheckActor { FlycheckActor { sender, config, workspace_root, check_process: None } } fn next_event(&self, inbox: &Receiver) -> Option { - let check_chan = self.check_process.as_ref().map(|(chan, _thread)| chan); + let check_chan = self.check_process.as_ref().map(|cargo| &cargo.receiver); select! { recv(inbox) -> msg => msg.ok().map(Event::Restart), recv(check_chan.unwrap_or(&never())) -> msg => Some(Event::CheckEvent(msg.ok())), @@ -166,7 +165,7 @@ impl FlycheckActor { self.send(Message::Progress(Progress::DidCancel)); } } - fn start_check_process(&self) -> (Receiver, jod_thread::JoinHandle) { + fn start_check_process(&self) -> CargoHandle { let mut cmd = match &self.config { FlycheckConfig::CargoCommand { command, @@ -199,32 +198,7 @@ impl FlycheckActor { }; cmd.current_dir(&self.workspace_root); - let (message_send, message_recv) = unbounded(); - let thread = 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 res = run_cargo(cmd, &mut |message| { - // Skip certain kinds of messages to only spend time on what's useful - match &message { - cargo_metadata::Message::CompilerArtifact(artifact) if artifact.fresh => { - return true - } - cargo_metadata::Message::BuildScriptExecuted(_) - | cargo_metadata::Message::Unknown => return true, - _ => {} - } - - // if the send channel was closed, we want to shutdown - message_send.send(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); - } - }); - (message_recv, thread) + CargoHandle::spawn(cmd) } fn send(&self, check_task: Message) { @@ -232,57 +206,90 @@ impl FlycheckActor { } } -fn run_cargo( - mut command: Command, - on_message: &mut dyn FnMut(cargo_metadata::Message) -> bool, -) -> io::Result<()> { - let child = - command.stdout(Stdio::piped()).stderr(Stdio::null()).stdin(Stdio::null()).spawn()?; - let mut child = ChildKiller(child); +struct CargoHandle { + receiver: Receiver, + #[allow(unused)] + thread: jod_thread::JoinHandle, +} - // 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 message in cargo_metadata::Message::parse_stream(stdout) { - let message = match message { - Ok(message) => message, - Err(err) => { - log::error!("Invalid json from cargo check, ignoring ({})", err); - continue; +impl CargoHandle { + fn spawn(command: Command) -> CargoHandle { + let (sender, receiver) = unbounded(); + let actor = CargoActor::new(command, sender); + let thread = jod_thread::spawn(move || { + let _ = actor.run(); + }); + CargoHandle { receiver, thread } + } +} + +struct CargoActor { + command: Command, + sender: Sender, +} + +impl CargoActor { + fn new(command: Command, sender: Sender) -> CargoActor { + CargoActor { command, sender } + } + fn run(mut self) -> io::Result<()> { + let child = self + .command + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .stdin(Stdio::null()) + .spawn()?; + let mut child = ChildKiller(child); + + // 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 message in cargo_metadata::Message::parse_stream(stdout) { + let message = match message { + Ok(message) => message, + Err(err) => { + log::error!("Invalid json from cargo check, ignoring ({})", err); + continue; + } + }; + + read_at_least_one_message = true; + + // Skip certain kinds of messages to only spend time on what's useful + match &message { + cargo_metadata::Message::CompilerArtifact(artifact) if artifact.fresh => continue, + cargo_metadata::Message::BuildScriptExecuted(_) + | cargo_metadata::Message::Unknown => continue, + _ => { + // if the send channel was closed, we want to shutdown + if self.sender.send(message).is_err() { + break; + } + } } - }; - - 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 exit_status = child.wait()?; + if !exit_status.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 + + // 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,the command produced no valid metadata (exit code: {:?}): {:?}", exit_status, self.command); + } + Ok(()) } - - // It is okay to ignore the result, as it only errors if the process is already dead - let _ = child.kill(); - - let exit_status = child.wait()?; - if !exit_status.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 - return Err(io::Error::new( - io::ErrorKind::Other, - format!( - "the command produced no valid metadata (exit code: {:?}): {:?}", - exit_status, command - ), - )); - } - - Ok(()) } struct ChildKiller(process::Child); From 5cdd8d442ef5a573f4af07e68dce7720ca603aba Mon Sep 17 00:00:00 2001 From: Aleksey Kladov Date: Sun, 28 Jun 2020 23:42:44 +0200 Subject: [PATCH 5/5] Cleanup cargo process handling in flycheck --- crates/flycheck/src/lib.rs | 121 +++++++++++++------------- crates/rust-analyzer/src/main_loop.rs | 6 +- 2 files changed, 66 insertions(+), 61 deletions(-) diff --git a/crates/flycheck/src/lib.rs b/crates/flycheck/src/lib.rs index ab1d71b981..1023d3040f 100644 --- a/crates/flycheck/src/lib.rs +++ b/crates/flycheck/src/lib.rs @@ -85,7 +85,7 @@ pub enum Message { pub enum Progress { DidStart, DidCheckCrate(String), - DidFinish, + DidFinish(io::Result<()>), DidCancel, } @@ -100,7 +100,7 @@ struct FlycheckActor { /// 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. - check_process: Option, + cargo_handle: Option, } enum Event { @@ -114,10 +114,10 @@ impl FlycheckActor { config: FlycheckConfig, workspace_root: PathBuf, ) -> FlycheckActor { - FlycheckActor { sender, config, workspace_root, check_process: None } + FlycheckActor { sender, config, workspace_root, cargo_handle: None } } fn next_event(&self, inbox: &Receiver) -> Option { - let check_chan = self.check_process.as_ref().map(|cargo| &cargo.receiver); + let check_chan = self.cargo_handle.as_ref().map(|cargo| &cargo.receiver); select! { recv(inbox) -> msg => msg.ok().map(Event::Restart), recv(check_chan.unwrap_or(&never())) -> msg => Some(Event::CheckEvent(msg.ok())), @@ -128,15 +128,22 @@ impl FlycheckActor { match event { Event::Restart(Restart) => { while let Ok(Restart) = inbox.recv_timeout(Duration::from_millis(50)) {} + self.cancel_check_process(); - self.check_process = Some(self.start_check_process()); - self.send(Message::Progress(Progress::DidStart)); + + let mut command = self.check_command(); + command.stdout(Stdio::piped()).stderr(Stdio::null()).stdin(Stdio::null()); + if let Ok(child) = command.spawn().map(JodChild) { + self.cargo_handle = Some(CargoHandle::spawn(child)); + self.send(Message::Progress(Progress::DidStart)); + } } Event::CheckEvent(None) => { // Watcher finished, replace it with a never channel to // avoid busy-waiting. - assert!(self.check_process.take().is_some()); - self.send(Message::Progress(Progress::DidFinish)); + let cargo_handle = self.cargo_handle.take().unwrap(); + let res = cargo_handle.join(); + self.send(Message::Progress(Progress::DidFinish(res))); } Event::CheckEvent(Some(message)) => match message { cargo_metadata::Message::CompilerArtifact(msg) => { @@ -161,11 +168,11 @@ impl FlycheckActor { self.cancel_check_process(); } fn cancel_check_process(&mut self) { - if self.check_process.take().is_some() { + if self.cargo_handle.take().is_some() { self.send(Message::Progress(Progress::DidCancel)); } } - fn start_check_process(&self) -> CargoHandle { + fn check_command(&self) -> Command { let mut cmd = match &self.config { FlycheckConfig::CargoCommand { command, @@ -197,8 +204,7 @@ impl FlycheckActor { } }; cmd.current_dir(&self.workspace_root); - - CargoHandle::spawn(cmd) + cmd } fn send(&self, check_task: Message) { @@ -207,49 +213,62 @@ impl FlycheckActor { } struct CargoHandle { - receiver: Receiver, + child: JodChild, #[allow(unused)] - thread: jod_thread::JoinHandle, + thread: jod_thread::JoinHandle>, + receiver: Receiver, } impl CargoHandle { - fn spawn(command: Command) -> CargoHandle { + fn spawn(mut child: JodChild) -> CargoHandle { + let child_stdout = child.stdout.take().unwrap(); let (sender, receiver) = unbounded(); - let actor = CargoActor::new(command, sender); - let thread = jod_thread::spawn(move || { - let _ = actor.run(); - }); - CargoHandle { receiver, thread } + let actor = CargoActor::new(child_stdout, sender); + let thread = jod_thread::spawn(move || actor.run()); + CargoHandle { child, thread, receiver } + } + fn join(mut self) -> io::Result<()> { + // It is okay to ignore the result, as it only errors if the process is already dead + let _ = self.child.kill(); + let exit_status = self.child.wait()?; + let read_at_least_one_message = self.thread.join()?; + if !exit_status.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 + return Err(io::Error::new( + io::ErrorKind::Other, + format!( + "Cargo watcher failed,the command produced no valid metadata (exit code: {:?})", + exit_status + ), + )); + } + Ok(()) } } struct CargoActor { - command: Command, + child_stdout: process::ChildStdout, sender: Sender, } impl CargoActor { - fn new(command: Command, sender: Sender) -> CargoActor { - CargoActor { command, sender } + fn new( + child_stdout: process::ChildStdout, + sender: Sender, + ) -> CargoActor { + CargoActor { child_stdout, sender } } - fn run(mut self) -> io::Result<()> { - let child = self - .command - .stdout(Stdio::piped()) - .stderr(Stdio::null()) - .stdin(Stdio::null()) - .spawn()?; - let mut child = ChildKiller(child); - + fn run(self) -> io::Result { // 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. + // be resilient 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 stdout = BufReader::new(self.child_stdout); let mut read_at_least_one_message = false; for message in cargo_metadata::Message::parse_stream(stdout) { let message = match message { @@ -264,50 +283,32 @@ impl CargoActor { // Skip certain kinds of messages to only spend time on what's useful match &message { - cargo_metadata::Message::CompilerArtifact(artifact) if artifact.fresh => continue, + cargo_metadata::Message::CompilerArtifact(artifact) if artifact.fresh => (), cargo_metadata::Message::BuildScriptExecuted(_) - | cargo_metadata::Message::Unknown => continue, - _ => { - // if the send channel was closed, we want to shutdown - if self.sender.send(message).is_err() { - break; - } - } + | cargo_metadata::Message::Unknown => (), + _ => self.sender.send(message).unwrap(), } } - - // It is okay to ignore the result, as it only errors if the process is already dead - let _ = child.kill(); - - let exit_status = child.wait()?; - if !exit_status.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 - - // 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,the command produced no valid metadata (exit code: {:?}): {:?}", exit_status, self.command); - } - Ok(()) + Ok(read_at_least_one_message) } } -struct ChildKiller(process::Child); +struct JodChild(process::Child); -impl ops::Deref for ChildKiller { +impl ops::Deref for JodChild { type Target = process::Child; fn deref(&self) -> &process::Child { &self.0 } } -impl ops::DerefMut for ChildKiller { +impl ops::DerefMut for JodChild { fn deref_mut(&mut self) -> &mut process::Child { &mut self.0 } } -impl Drop for ChildKiller { +impl Drop for JodChild { fn drop(&mut self) { let _ = self.0.kill(); } diff --git a/crates/rust-analyzer/src/main_loop.rs b/crates/rust-analyzer/src/main_loop.rs index e5194fe41a..9fd16ef3b4 100644 --- a/crates/rust-analyzer/src/main_loop.rs +++ b/crates/rust-analyzer/src/main_loop.rs @@ -216,7 +216,11 @@ impl GlobalState { flycheck::Progress::DidCheckCrate(target) => { (Progress::Report, Some(target)) } - flycheck::Progress::DidFinish | flycheck::Progress::DidCancel => { + flycheck::Progress::DidCancel => (Progress::End, None), + flycheck::Progress::DidFinish(result) => { + if let Err(err) = result { + log::error!("cargo check failed: {}", err) + } (Progress::End, None) } };