mikros/rust
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Auto merge of #12520 - Veykril:flycheck-cancel, r=Veykril

Browse Source 
internal: Bring back JodChild into flychecking for cancellation

cc https://github.com/rust-lang/rust-analyzer/pull/10517/files#r895241975
This commit is contained in:
bors 2022-06-24 17:42:00 +00:00
commit 5bb123d970
4 changed files with 134 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

105
crates/flycheck/src/lib.rs
View File

@ -2,12 +2,16 @@
//! another compatible command (f.x. clippy) in a background thread and provide
//! LSP diagnostics based on the output of the command.
use std::{fmt, io, process::Command, time::Duration};
use std::{
fmt, io,
process::{ChildStderr, ChildStdout, Command, Stdio},
time::Duration,
};
use crossbeam_channel::{never, select, unbounded, Receiver, Sender};
use paths::AbsPathBuf;
use serde::Deserialize;
use stdx::process::streaming_output;
use stdx::{process::streaming_output, JodChild};
pub use cargo_metadata::diagnostic::{
Applicability, Diagnostic, DiagnosticCode, DiagnosticLevel, DiagnosticSpan,
@ -117,7 +121,7 @@ struct FlycheckActor {
sender: Box<dyn Fn(Message) + Send>,
config: FlycheckConfig,
workspace_root: AbsPathBuf,
/// WatchThread exists to wrap around the communication needed to be able to
/// CargoHandle 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
@ -153,18 +157,36 @@ impl FlycheckActor {
while let Some(event) = self.next_event(&inbox) {
match event {
Event::Restart(Restart) => {
if let Some(cargo_handle) = self.cargo_handle.take() {
// Cancel the previously spawned process
cargo_handle.cancel();
}
while let Ok(Restart) = inbox.recv_timeout(Duration::from_millis(50)) {}
self.cancel_check_process();
self.progress(Progress::DidCancel);
let command = self.check_command();
tracing::info!("restart flycheck {:?}", command);
self.cargo_handle = Some(CargoHandle::spawn(command));
self.progress(Progress::DidStart);
tracing::debug!(?command, "will restart flycheck");
match CargoHandle::spawn(command) {
Ok(cargo_handle) => {
tracing::debug!(
command = ?self.check_command(),
"did restart flycheck"
);
self.cargo_handle = Some(cargo_handle);
self.progress(Progress::DidStart);
}
Err(error) => {
tracing::error!(
command = ?self.check_command(),
%error, "failed to restart flycheck"
);
}
}
}
Event::CheckEvent(None) => {
// Watcher finished, replace it with a never channel to
// avoid busy-waiting.
tracing::debug!("flycheck finished");
// Watcher finished
let cargo_handle = self.cargo_handle.take().unwrap();
let res = cargo_handle.join();
if res.is_err() {
@ -192,8 +214,10 @@ impl FlycheckActor {
// If we rerun the thread, we need to discard the previous check results first
self.cancel_check_process();
}
fn cancel_check_process(&mut self) {
if self.cargo_handle.take().is_some() {
if let Some(cargo_handle) = self.cargo_handle.take() {
cargo_handle.cancel();
self.progress(Progress::DidCancel);
}
}
@ -249,37 +273,64 @@ impl FlycheckActor {
}
}
/// A handle to a cargo process used for fly-checking.
struct CargoHandle {
thread: jod_thread::JoinHandle<io::Result<()>>,
/// The handle to the actual cargo process. As we cannot cancel directly from with
/// a read syscall dropping and therefor terminating the process is our best option.
child: JodChild,
thread: jod_thread::JoinHandle<io::Result<(bool, String)>>,
receiver: Receiver<CargoMessage>,
}
impl CargoHandle {
fn spawn(command: Command) -> CargoHandle {
fn spawn(mut command: Command) -> std::io::Result<CargoHandle> {
command.stdout(Stdio::piped()).stderr(Stdio::piped()).stdin(Stdio::null());
let mut child = JodChild::spawn(command)?;
let stdout = child.stdout.take().unwrap();
let stderr = child.stderr.take().unwrap();
let (sender, receiver) = unbounded();
let actor = CargoActor::new(sender);
let actor = CargoActor::new(sender, stdout, stderr);
let thread = jod_thread::Builder::new()
.name("CargoHandle".to_owned())
.spawn(move || actor.run(command))
.spawn(move || actor.run())
.expect("failed to spawn thread");
CargoHandle { thread, receiver }
Ok(CargoHandle { child, thread, receiver })
}
fn join(self) -> io::Result<()> {
self.thread.join()
fn cancel(mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
fn join(mut self) -> io::Result<()> {
let _ = self.child.kill();
let exit_status = self.child.wait()?;
let (read_at_least_one_message, error) = self.thread.join()?;
if read_at_least_one_message || exit_status.success() {
Ok(())
} else {
Err(io::Error::new(io::ErrorKind::Other, format!(
"Cargo watcher failed, the command produced no valid metadata (exit code: {:?}):\n{}",
exit_status, error
)))
}
}
}
struct CargoActor {
sender: Sender<CargoMessage>,
stdout: ChildStdout,
stderr: ChildStderr,
}
impl CargoActor {
fn new(sender: Sender<CargoMessage>) -> CargoActor {
CargoActor { sender }
fn new(sender: Sender<CargoMessage>, stdout: ChildStdout, stderr: ChildStderr) -> CargoActor {
CargoActor { sender, stdout, stderr }
}
fn run(self, command: Command) -> io::Result<()> {
fn run(self) -> io::Result<(bool, String)> {
// 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
@ -292,7 +343,8 @@ impl CargoActor {
let mut error = String::new();
let mut read_at_least_one_message = false;
let output = streaming_output(
command,
self.stdout,
self.stderr,
&mut |line| {
read_at_least_one_message = true;
@ -325,14 +377,7 @@ impl CargoActor {
},
);
match output {
Ok(_) if read_at_least_one_message => Ok(()),
Ok(output) if output.status.success() => Ok(()),
Ok(output) => {
Err(io::Error::new(io::ErrorKind::Other, format!(
"Cargo watcher failed, the command produced no valid metadata (exit code: {:?}):\n{}",
output.status, error
)))
}
Ok(_) => Ok((read_at_least_one_message, error)),
Err(e) => Err(io::Error::new(e.kind(), format!("{:?}: {}", e, error))),
}
}

2
crates/project-model/src/build_scripts.rs
View File

@ -110,7 +110,7 @@ impl WorkspaceBuildScripts {
};
tracing::info!("Running build scripts: {:?}", cmd);
let output = stdx::process::streaming_output(
let output = stdx::process::spawn_with_streaming_output(
cmd,
&mut |line| {
// Copy-pasted from existing cargo_metadata. It seems like we

8
crates/stdx/src/lib.rs
View File

@ -1,6 +1,7 @@
//! Missing batteries for standard libraries.
use std::iter;
use std::process::Command;
use std::{cmp::Ordering, ops, time::Instant};
use std::{io as sio, iter};
mod macros;
pub mod process;
@ -132,6 +133,7 @@ pub fn defer<F: FnOnce()>(f: F) -> impl Drop {
D(Some(f))
}
/// A [`std::process::Child`] wrapper that will kill the child on drop.
#[cfg_attr(not(target_arch = "wasm32"), repr(transparent))]
#[derive(Debug)]
pub struct JodChild(pub std::process::Child);
@ -157,6 +159,10 @@ impl Drop for JodChild {
}
impl JodChild {
pub fn spawn(mut command: Command) -> sio::Result<Self> {
command.spawn().map(Self)
}
pub fn into_inner(self) -> std::process::Child {
if cfg!(target_arch = "wasm32") {
panic!("no processes on wasm");

90
crates/stdx/src/process.rs
View File

@ -5,54 +5,66 @@
use std::{
io,
process::{Command, Output, Stdio},
process::{ChildStderr, ChildStdout, Command, Output, Stdio},
};
use crate::JodChild;
pub fn streaming_output(
out: ChildStdout,
err: ChildStderr,
on_stdout_line: &mut dyn FnMut(&str),
on_stderr_line: &mut dyn FnMut(&str),
) -> io::Result<(Vec<u8>, Vec<u8>)> {
let mut stdout = Vec::new();
let mut stderr = Vec::new();
imp::read2(out, err, &mut |is_out, data, eof| {
let idx = if eof {
data.len()
} else {
match data.iter().rposition(|b| *b == b'\n') {
Some(i) => i + 1,
None => return,
}
};
{
// scope for new_lines
let new_lines = {
let dst = if is_out { &mut stdout } else { &mut stderr };
let start = dst.len();
let data = data.drain(..idx);
dst.extend(data);
&dst[start..]
};
for line in String::from_utf8_lossy(new_lines).lines() {
if is_out {
on_stdout_line(line);
} else {
on_stderr_line(line);
}
}
}
})?;
Ok((stdout, stderr))
}
pub fn spawn_with_streaming_output(
mut cmd: Command,
on_stdout_line: &mut dyn FnMut(&str),
on_stderr_line: &mut dyn FnMut(&str),
) -> io::Result<Output> {
let mut stdout = Vec::new();
let mut stderr = Vec::new();
let cmd = cmd.stdout(Stdio::piped()).stderr(Stdio::piped()).stdin(Stdio::null());
let status = {
let mut child = cmd.spawn()?;
let out = child.stdout.take().unwrap();
let err = child.stderr.take().unwrap();
imp::read2(out, err, &mut |is_out, data, eof| {
let idx = if eof {
data.len()
} else {
match data.iter().rposition(|b| *b == b'\n') {
Some(i) => i + 1,
None => return,
}
};
{
// scope for new_lines
let new_lines = {
let dst = if is_out { &mut stdout } else { &mut stderr };
let start = dst.len();
let data = data.drain(..idx);
dst.extend(data);
&dst[start..]
};
for line in String::from_utf8_lossy(new_lines).lines() {
if is_out {
on_stdout_line(line);
} else {
on_stderr_line(line);
}
}
}
})?;
let _ = child.kill();
child.wait()?
};
let mut child = JodChild(cmd.spawn()?);
let (stdout, stderr) = streaming_output(
child.stdout.take().unwrap(),
child.stderr.take().unwrap(),
on_stdout_line,
on_stderr_line,
)?;
let status = child.wait()?;
Ok(Output { status, stdout, stderr })
}