From 27c15471c1329f287d47e3a79ccc7a3cd185315b Mon Sep 17 00:00:00 2001 From: Ivan Petkov Date: Mon, 24 Jun 2019 17:19:16 -0700 Subject: [PATCH] process: Run cargo fmt --- src/lib.rs | 56 +++++++++++++++++------------------- src/unix/mod.rs | 68 ++++++++++++++++++++++--------------------- src/unix/orphan.rs | 6 ++-- src/unix/reap.rs | 72 +++++++++++++++++++++------------------------- src/windows.rs | 40 +++++++++++++------------- tests/issue_42.rs | 28 ++++++++++-------- tests/smoke.rs | 3 +- tests/stdio.rs | 33 ++++++++++----------- 8 files changed, 151 insertions(+), 155 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 7464b69d7..5af9f68a2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -172,12 +172,12 @@ extern crate log; use std::io::{self, Read, Write}; use std::process::{Command, ExitStatus, Output, Stdio}; -use futures::{Async, Future, Poll, IntoFuture}; -use futures::future::{Either, ok}; +use futures::future::{ok, Either}; +use futures::{Async, Future, IntoFuture, Poll}; use kill::Kill; use std::fmt; -use tokio_io::io::{read_to_end}; -use tokio_io::{AsyncWrite, AsyncRead, IoFuture}; +use tokio_io::io::read_to_end; +use tokio_io::{AsyncRead, AsyncWrite, IoFuture}; use tokio_reactor::Handle; #[path = "unix/mod.rs"] @@ -337,13 +337,12 @@ struct SpawnedChild { impl CommandExt for Command { fn spawn_async_with_handle(&mut self, handle: &Handle) -> io::Result { - imp::spawn_child(self, handle) - .map(|spawned_child| Child { - child: ChildDropGuard::new(spawned_child.child), - stdin: spawned_child.stdin.map(|inner| ChildStdin { inner }), - stdout: spawned_child.stdout.map(|inner| ChildStdout { inner }), - stderr: spawned_child.stderr.map(|inner| ChildStderr { inner }), - }) + imp::spawn_child(self, handle).map(|spawned_child| Child { + child: ChildDropGuard::new(spawned_child.child), + stdin: spawned_child.stdin.map(|inner| ChildStdin { inner }), + stdout: spawned_child.stdout.map(|inner| ChildStdout { inner }), + stderr: spawned_child.stderr.map(|inner| ChildStderr { inner }), + }) } fn status_async_with_handle(&mut self, handle: &Handle) -> io::Result { @@ -355,9 +354,7 @@ impl CommandExt for Command { child.stdout.take(); child.stderr.take(); - StatusAsync { - inner: child, - } + StatusAsync { inner: child } }) } @@ -365,7 +362,8 @@ impl CommandExt for Command { self.stdout(Stdio::piped()); self.stderr(Stdio::piped()); - let inner = self.spawn_async_with_handle(handle) + let inner = self + .spawn_async_with_handle(handle) .into_future() .and_then(Child::wait_with_output); @@ -416,7 +414,6 @@ impl Drop for ChildDropGuard { } } - impl Future for ChildDropGuard { type Item = T::Item; type Error = T::Error; @@ -514,13 +511,14 @@ impl Child { }; WaitWithOutput { - inner: Box::new(self.join3(stdout, stderr).map(|(status, stdout, stderr)| { - Output { - status, - stdout, - stderr, - } - })) + inner: Box::new( + self.join3(stdout, stderr) + .map(|(status, stdout, stderr)| Output { + status, + stdout, + stderr, + }), + ), } } @@ -700,8 +698,7 @@ impl Read for ChildStdout { } } -impl AsyncRead for ChildStdout { -} +impl AsyncRead for ChildStdout {} impl Read for ChildStderr { fn read(&mut self, bytes: &mut [u8]) -> io::Result { @@ -709,13 +706,12 @@ impl Read for ChildStderr { } } -impl AsyncRead for ChildStderr { -} +impl AsyncRead for ChildStderr {} #[cfg(unix)] mod sys { + use super::{ChildStderr, ChildStdin, ChildStdout}; use std::os::unix::io::{AsRawFd, RawFd}; - use super::{ChildStdin, ChildStdout, ChildStderr}; impl AsRawFd for ChildStdin { fn as_raw_fd(&self) -> RawFd { @@ -738,8 +734,8 @@ mod sys { #[cfg(windows)] mod sys { + use super::{ChildStderr, ChildStdin, ChildStdout}; use std::os::windows::io::{AsRawHandle, RawHandle}; - use super::{ChildStdin, ChildStdout, ChildStderr}; impl AsRawHandle for ChildStdin { fn as_raw_handle(&self) -> RawHandle { @@ -762,10 +758,10 @@ mod sys { #[cfg(test)] mod test { + use super::ChildDropGuard; use futures::{Async, Future, Poll}; use kill::Kill; use std::io; - use super::ChildDropGuard; struct Mock { num_kills: usize, diff --git a/src/unix/mod.rs b/src/unix/mod.rs index 465117d3c..ad6f97d76 100644 --- a/src/unix/mod.rs +++ b/src/unix/mod.rs @@ -28,20 +28,20 @@ extern crate tokio_signal; mod orphan; mod reap; -use futures::future::FlattenStream; -use futures::{Future, Poll}; -use kill::Kill; -use self::mio::{Poll as MioPoll, PollOpt, Ready, Token}; -use self::mio::unix::{EventedFd, UnixReady}; use self::mio::event::Evented; +use self::mio::unix::{EventedFd, UnixReady}; +use self::mio::{Poll as MioPoll, PollOpt, Ready, Token}; use self::orphan::{AtomicOrphanQueue, OrphanQueue, Wait}; use self::reap::Reaper; use self::tokio_signal::unix::Signal; +use super::SpawnedChild; +use futures::future::FlattenStream; +use futures::{Future, Poll}; +use kill::Kill; use std::fmt; use std::io; use std::os::unix::io::{AsRawFd, RawFd}; use std::process::{self, ExitStatus}; -use super::SpawnedChild; use tokio_io::IoFuture; use tokio_reactor::{Handle, PollEvented}; @@ -153,7 +153,10 @@ impl io::Write for Fd { } } -impl AsRawFd for Fd where T: AsRawFd { +impl AsRawFd for Fd +where + T: AsRawFd, +{ fn as_raw_fd(&self) -> RawFd { self.0.as_raw_fd() } @@ -163,29 +166,28 @@ pub type ChildStdin = PollEvented>; pub type ChildStdout = PollEvented>; pub type ChildStderr = PollEvented>; -impl Evented for Fd where T: AsRawFd { - fn register(&self, - poll: &MioPoll, - token: Token, - interest: Ready, - opts: PollOpt) - -> io::Result<()> { - EventedFd(&self.as_raw_fd()).register(poll, - token, - interest | UnixReady::hup(), - opts) +impl Evented for Fd +where + T: AsRawFd, +{ + fn register( + &self, + poll: &MioPoll, + token: Token, + interest: Ready, + opts: PollOpt, + ) -> io::Result<()> { + EventedFd(&self.as_raw_fd()).register(poll, token, interest | UnixReady::hup(), opts) } - fn reregister(&self, - poll: &MioPoll, - token: Token, - interest: Ready, - opts: PollOpt) - -> io::Result<()> { - EventedFd(&self.as_raw_fd()).reregister(poll, - token, - interest | UnixReady::hup(), - opts) + fn reregister( + &self, + poll: &MioPoll, + token: Token, + interest: Ready, + opts: PollOpt, + ) -> io::Result<()> { + EventedFd(&self.as_raw_fd()).reregister(poll, token, interest | UnixReady::hup(), opts) } fn deregister(&self, poll: &MioPoll) -> io::Result<()> { @@ -193,9 +195,9 @@ impl Evented for Fd where T: AsRawFd { } } -fn stdio(option: Option, handle: &Handle) - -> io::Result>>> - where T: AsRawFd +fn stdio(option: Option, handle: &Handle) -> io::Result>>> +where + T: AsRawFd, { let io = match option { Some(io) => io, @@ -207,11 +209,11 @@ fn stdio(option: Option, handle: &Handle) let fd = io.as_raw_fd(); let r = libc::fcntl(fd, libc::F_GETFL); if r == -1 { - return Err(io::Error::last_os_error()) + return Err(io::Error::last_os_error()); } let r = libc::fcntl(fd, libc::F_SETFL, r | libc::O_NONBLOCK); if r == -1 { - return Err(io::Error::last_os_error()) + return Err(io::Error::last_os_error()); } } let io = try!(PollEvented::new_with_handle(Fd(io), handle)); diff --git a/src/unix/orphan.rs b/src/unix/orphan.rs index 6b6a2f287..6f456ce0e 100644 --- a/src/unix/orphan.rs +++ b/src/unix/orphan.rs @@ -70,7 +70,7 @@ impl OrphanQueue for AtomicOrphanQueue { let mut orphans = Vec::with_capacity(len); while let Ok(mut orphan) = self.queue.pop() { match orphan.try_wait() { - Ok(Some(_)) => {}, + Ok(Some(_)) => {} Err(e) => error!( "leaking orphaned process {} due to try_wait() error: {}", orphan.id(), @@ -92,13 +92,13 @@ impl OrphanQueue for AtomicOrphanQueue { #[cfg(test)] mod test { + use super::Wait; + use super::{AtomicOrphanQueue, OrphanQueue}; use std::cell::Cell; use std::io; use std::os::unix::process::ExitStatusExt; use std::process::ExitStatus; use std::rc::Rc; - use super::{AtomicOrphanQueue, OrphanQueue}; - use super::Wait; struct MockWait { total_waits: Rc>, diff --git a/src/unix/reap.rs b/src/unix/reap.rs index 76e995c1b..567319b36 100644 --- a/src/unix/reap.rs +++ b/src/unix/reap.rs @@ -1,16 +1,17 @@ +use super::orphan::{OrphanQueue, Wait}; use futures::{Async, Future, Poll, Stream}; use kill::Kill; use std::io; use std::ops::Deref; use std::process::ExitStatus; -use super::orphan::{OrphanQueue, Wait}; /// Orchestrates between registering interest for receiving signals when a /// child process has exited, and attempting to poll for process completion. #[derive(Debug)] pub(crate) struct Reaper - where W: Wait, - Q: OrphanQueue, +where + W: Wait, + Q: OrphanQueue, { inner: Option, orphan_queue: Q, @@ -18,8 +19,9 @@ pub(crate) struct Reaper } impl Deref for Reaper - where W: Wait, - Q: OrphanQueue, +where + W: Wait, + Q: OrphanQueue, { type Target = W; @@ -29,8 +31,9 @@ impl Deref for Reaper } impl Reaper - where W: Wait, - Q: OrphanQueue, +where + W: Wait, + Q: OrphanQueue, { pub(crate) fn new(inner: W, orphan_queue: Q, signal: S) -> Self { Self { @@ -50,9 +53,10 @@ impl Reaper } impl Future for Reaper - where W: Wait, - Q: OrphanQueue, - S: Stream, +where + W: Wait, + Q: OrphanQueue, + S: Stream, { type Item = ExitStatus; type Error = io::Error; @@ -100,18 +104,19 @@ impl Future for Reaper } impl Kill for Reaper - where W: Kill + Wait, - Q: OrphanQueue, +where + W: Kill + Wait, + Q: OrphanQueue, { fn kill(&mut self) -> io::Result<()> { self.inner_mut().kill() } } - impl Drop for Reaper - where W: Wait, - Q: OrphanQueue, +where + W: Wait, + Q: OrphanQueue, { fn drop(&mut self) { if let Ok(Some(_)) = self.inner_mut().try_wait() { @@ -125,11 +130,11 @@ impl Drop for Reaper #[cfg(test)] mod test { + use super::*; use futures::{Async, Poll, Stream}; use std::cell::{Cell, RefCell}; - use std::process::ExitStatus; use std::os::unix::process::ExitStatusExt; - use super::*; + use std::process::ExitStatus; #[derive(Debug)] struct MockWait { @@ -145,7 +150,7 @@ mod test { total_kills: 0, total_waits: 0, num_wait_until_status, - status + status, } } } @@ -183,7 +188,7 @@ mod test { fn new(values: Vec>) -> Self { Self { total_polls: 0, - values + values, } } } @@ -217,8 +222,7 @@ mod test { impl OrphanQueue for MockQueue { fn push_orphan(&self, orphan: W) { - self.all_enqueued.borrow_mut() - .push(orphan); + self.all_enqueued.borrow_mut().push(orphan); } fn reap_orphans(&self) { @@ -230,13 +234,11 @@ mod test { fn reaper() { let exit = ExitStatus::from_raw(0); let mock = MockWait::new(exit, 3); - let mut grim = Reaper::new(mock, MockQueue::new(), MockStream::new(vec!( - None, - Some(()), - None, - None, - None, - ))); + let mut grim = Reaper::new( + mock, + MockQueue::new(), + MockStream::new(vec![None, Some(()), None, None, None]), + ); // Not yet exited, interest registered assert_eq!(Async::NotReady, grim.poll().expect("failed to wait")); @@ -267,7 +269,7 @@ mod test { let mut grim = Reaper::new( MockWait::new(exit, 0), MockQueue::new(), - MockStream::new(vec!(None)) + MockStream::new(vec![None]), ); grim.kill().unwrap(); @@ -284,11 +286,7 @@ mod test { { let queue = MockQueue::new(); - let grim = Reaper::new( - &mut mock, - &queue, - MockStream::new(vec!()) - ); + let grim = Reaper::new(&mut mock, &queue, MockStream::new(vec![])); drop(grim); @@ -307,11 +305,7 @@ mod test { { let queue = MockQueue::<&mut MockWait>::new(); - let grim = Reaper::new( - &mut mock, - &queue, - MockStream::new(vec!()) - ); + let grim = Reaper::new(&mut mock, &queue, MockStream::new(vec![])); drop(grim); assert_eq!(0, queue.total_reaps.get()); diff --git a/src/windows.rs b/src/windows.rs index 253f35476..a368d510b 100644 --- a/src/windows.rs +++ b/src/windows.rs @@ -15,8 +15,8 @@ //! `RegisterWaitForSingleObject` and then wait on the other end of the oneshot //! from then on out. -extern crate winapi; extern crate mio_named_pipes; +extern crate winapi; use std::fmt; use std::io; @@ -25,10 +25,6 @@ use std::os::windows::process::ExitStatusExt; use std::process::{self, ExitStatus}; use std::ptr; -use futures::future::Fuse; -use futures::sync::oneshot; -use futures::{Future, Poll, Async}; -use kill::Kill; use self::mio_named_pipes::NamedPipe; use self::winapi::shared::minwindef::*; use self::winapi::shared::winerror::*; @@ -39,6 +35,10 @@ use self::winapi::um::threadpoollegacyapiset::*; use self::winapi::um::winbase::*; use self::winapi::um::winnt::*; use super::SpawnedChild; +use futures::future::Fuse; +use futures::sync::oneshot; +use futures::{Async, Future, Poll}; +use kill::Kill; use tokio_reactor::{Handle, PollEvented}; #[must_use = "futures do nothing unless polled"] @@ -107,28 +107,29 @@ impl Future for Child { Async::NotReady => return Ok(Async::NotReady), } let status = try!(try_wait(&self.child)).expect("not ready yet"); - return Ok(status.into()) + return Ok(status.into()); } if let Some(e) = try!(try_wait(&self.child)) { - return Ok(e.into()) + return Ok(e.into()); } let (tx, rx) = oneshot::channel(); let ptr = Box::into_raw(Box::new(Some(tx))); let mut wait_object = ptr::null_mut(); let rc = unsafe { - RegisterWaitForSingleObject(&mut wait_object, - self.child.as_raw_handle(), - Some(callback), - ptr as *mut _, - INFINITE, - WT_EXECUTEINWAITTHREAD | - WT_EXECUTEONLYONCE) + RegisterWaitForSingleObject( + &mut wait_object, + self.child.as_raw_handle(), + Some(callback), + ptr as *mut _, + INFINITE, + WT_EXECUTEINWAITTHREAD | WT_EXECUTEONLYONCE, + ) }; if rc == 0 { let err = io::Error::last_os_error(); drop(unsafe { Box::from_raw(ptr) }); - return Err(err) + return Err(err); } self.waiting = Some(Waiting { rx: rx.fuse(), @@ -151,8 +152,7 @@ impl Drop for Waiting { } } -unsafe extern "system" fn callback(ptr: PVOID, - _timer_fired: BOOLEAN) { +unsafe extern "system" fn callback(ptr: PVOID, _timer_fired: BOOLEAN) { let complete = &mut *(ptr as *mut Option>); let _ = complete.take().unwrap().send(()); } @@ -178,9 +178,9 @@ pub type ChildStdin = PollEvented; pub type ChildStdout = PollEvented; pub type ChildStderr = PollEvented; -fn stdio(option: Option, handle: &Handle) - -> io::Result>> - where T: IntoRawHandle, +fn stdio(option: Option, handle: &Handle) -> io::Result>> +where + T: IntoRawHandle, { let io = match option { Some(io) => io, diff --git a/tests/issue_42.rs b/tests/issue_42.rs index a0a9700c2..0b34efcf1 100644 --- a/tests/issue_42.rs +++ b/tests/issue_42.rs @@ -3,10 +3,10 @@ extern crate futures; extern crate tokio_process; -use futures::{Future, IntoFuture, Stream, stream}; +use futures::{stream, Future, IntoFuture, Stream}; use std::process::{Command, Stdio}; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use std::thread; use std::time::Duration; use tokio_process::CommandExt; @@ -17,15 +17,16 @@ fn run_test() { thread::spawn(move || { let _ = stream::iter_ok(0..2) - .map(|i| Command::new("echo") - .arg(format!("I am spawned process #{}", i)) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn_async() - .into_future() - .flatten() - ) + .map(|i| { + Command::new("echo") + .arg(format!("I am spawned process #{}", i)) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn_async() + .into_future() + .flatten() + }) .buffered(2) .collect() .wait(); @@ -34,7 +35,10 @@ fn run_test() { }); thread::sleep(Duration::from_millis(100)); - assert!(finished.load(Ordering::SeqCst), "FINISHED flag not set, maybe we deadlocked?"); + assert!( + finished.load(Ordering::SeqCst), + "FINISHED flag not set, maybe we deadlocked?" + ); } #[test] diff --git a/tests/smoke.rs b/tests/smoke.rs index a1dc7ec52..9d11002ad 100644 --- a/tests/smoke.rs +++ b/tests/smoke.rs @@ -14,8 +14,7 @@ fn simple() { let id = child.id(); assert!(id > 0); - let status = support::run_with_timeout(&mut child) - .expect("failed to run future"); + let status = support::run_with_timeout(&mut child).expect("failed to run future"); assert_eq!(status.code(), Some(2)); assert_eq!(child.id(), id); diff --git a/tests/stdio.rs b/tests/stdio.rs index 0f3a77b34..647350f72 100644 --- a/tests/stdio.rs +++ b/tests/stdio.rs @@ -5,19 +5,18 @@ extern crate tokio_io; extern crate tokio_process; use std::io; -use std::process::{Stdio, ExitStatus, Command}; +use std::process::{Command, ExitStatus, Stdio}; use futures::future::Future; use futures::stream::{self, Stream}; use tokio_io::io::{read_until, write_all}; -use tokio_process::{CommandExt, Child}; +use tokio_process::{Child, CommandExt}; mod support; fn cat() -> Command { let mut cmd = support::cmd("cat"); - cmd.stdin(Stdio::piped()) - .stdout(Stdio::piped()); + cmd.stdin(Stdio::piped()).stdout(Stdio::piped()); cmd } @@ -28,10 +27,12 @@ fn feed_cat(mut cat: Child, n: usize) -> Box Box= n; debug!("starting read from child"); read_until(reader, b'\n', Vec::new()).and_then(move |(reader, vec)| { - debug!("read line {} from child ({} bytes, done: {})", - i, vec.len(), done); + debug!( + "read line {} from child ({} bytes, done: {})", + i, + vec.len(), + done + ); match (done, vec.len()) { - (false, 0) => { - Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")) - }, - (true, n) if n != 0 => { - Err(io::Error::new(io::ErrorKind::Other, "extraneous data")) - }, + (false, 0) => Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")), + (true, n) if n != 0 => Err(io::Error::new(io::ErrorKind::Other, "extraneous data")), _ => { let s = std::str::from_utf8(&vec).unwrap(); let expected = format!("line {}\n", i);