tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-22 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

process: Run cargo fmt

Browse Source
This commit is contained in:
Ivan Petkov
2019-06-24 17:29:33 -07:00
parent 0ab25878bd
commit 27c15471c1
8 changed files with 151 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib.rs
+26 -30
View File
@@ -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<Child> {
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<StatusAsync> {
@@ -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<T: Kill> Drop for ChildDropGuard<T> {
}
}
impl<T: Future + Kill> Future for ChildDropGuard<T> {
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<usize> {
@@ -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,
src/unix/mod.rs
+35 -33
View File
@@ -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<T: io::Write> io::Write for Fd<T> {
}
}
impl<T> AsRawFd for Fd<T> where T: AsRawFd {
impl<T> AsRawFd for Fd<T>
where
T: AsRawFd,
{
fn as_raw_fd(&self) -> RawFd {
self.0.as_raw_fd()
}
@@ -163,29 +166,28 @@ pub type ChildStdin = PollEvented<Fd<process::ChildStdin>>;
pub type ChildStdout = PollEvented<Fd<process::ChildStdout>>;
pub type ChildStderr = PollEvented<Fd<process::ChildStderr>>;
impl<T> Evented for Fd<T> 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<T> Evented for Fd<T>
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<T> Evented for Fd<T> where T: AsRawFd {
}
}
fn stdio<T>(option: Option<T>, handle: &Handle)
-> io::Result<Option<PollEvented<Fd<T>>>>
where T: AsRawFd
fn stdio<T>(option: Option<T>, handle: &Handle) -> io::Result<Option<PollEvented<Fd<T>>>>
where
T: AsRawFd,
{
let io = match option {
Some(io) => io,
@@ -207,11 +209,11 @@ fn stdio<T>(option: Option<T>, 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));
src/unix/orphan.rs
+3 -3
View File
@@ -70,7 +70,7 @@ impl<T: Wait> OrphanQueue<T> for AtomicOrphanQueue<T> {
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<T: Wait> OrphanQueue<T> for AtomicOrphanQueue<T> {
#[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<Cell<usize>>,
src/unix/reap.rs
+33 -39
View File
@@ -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<W, Q, S>
where W: Wait,
Q: OrphanQueue<W>,
where
W: Wait,
Q: OrphanQueue<W>,
{
inner: Option<W>,
orphan_queue: Q,
@@ -18,8 +19,9 @@ pub(crate) struct Reaper<W, Q, S>
}
impl<W, Q, S> Deref for Reaper<W, Q, S>
where W: Wait,
Q: OrphanQueue<W>,
where
W: Wait,
Q: OrphanQueue<W>,
{
type Target = W;
@@ -29,8 +31,9 @@ impl<W, Q, S> Deref for Reaper<W, Q, S>
}
impl<W, Q, S> Reaper<W, Q, S>
where W: Wait,
Q: OrphanQueue<W>,
where
W: Wait,
Q: OrphanQueue<W>,
{
pub(crate) fn new(inner: W, orphan_queue: Q, signal: S) -> Self {
Self {
@@ -50,9 +53,10 @@ impl<W, Q, S> Reaper<W, Q, S>
}
impl<W, Q, S> Future for Reaper<W, Q, S>
where W: Wait,
Q: OrphanQueue<W>,
S: Stream<Error = io::Error>,
where
W: Wait,
Q: OrphanQueue<W>,
S: Stream<Error = io::Error>,
{
type Item = ExitStatus;
type Error = io::Error;
@@ -100,18 +104,19 @@ impl<W, Q, S> Future for Reaper<W, Q, S>
}
impl<W, Q, S> Kill for Reaper<W, Q, S>
where W: Kill + Wait,
Q: OrphanQueue<W>,
where
W: Kill + Wait,
Q: OrphanQueue<W>,
{
fn kill(&mut self) -> io::Result<()> {
self.inner_mut().kill()
}
}
impl<W, Q, S> Drop for Reaper<W, Q, S>
where W: Wait,
Q: OrphanQueue<W>,
where
W: Wait,
Q: OrphanQueue<W>,
{
fn drop(&mut self) {
if let Ok(Some(_)) = self.inner_mut().try_wait() {
@@ -125,11 +130,11 @@ impl<W, Q, S> Drop for Reaper<W, Q, S>
#[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<Option<()>>) -> Self {
Self {
total_polls: 0,
values
values,
}
}
}
@@ -217,8 +222,7 @@ mod test {
impl<W: Wait> OrphanQueue<W> for MockQueue<W> {
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());
src/windows.rs
+20 -20
View File
@@ -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<oneshot::Sender<()>>);
let _ = complete.take().unwrap().send(());
}
@@ -178,9 +178,9 @@ pub type ChildStdin = PollEvented<NamedPipe>;
pub type ChildStdout = PollEvented<NamedPipe>;
pub type ChildStderr = PollEvented<NamedPipe>;
fn stdio<T>(option: Option<T>, handle: &Handle)
-> io::Result<Option<PollEvented<NamedPipe>>>
where T: IntoRawHandle,
fn stdio<T>(option: Option<T>, handle: &Handle) -> io::Result<Option<PollEvented<NamedPipe>>>
where
T: IntoRawHandle,
{
let io = match option {
Some(io) => io,
tests/issue_42.rs
+16 -12
View File
@@ -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]
tests/smoke.rs
+1 -2
View File
@@ -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);
tests/stdio.rs
+17 -16
View File
@@ -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<Future<Item = ExitStatus, Error = i
debug!("starting to feed");
// Produce n lines on the child's stdout.
let numbers = stream::iter_ok(0..n);
let write = numbers.fold(stdin, |stdin, i| {
debug!("sending line {} to child", i);
write_all(stdin, format!("line {}\n", i).into_bytes()).map(|p| p.0)
}).map(|_| ());
let write = numbers
.fold(stdin, |stdin, i| {
debug!("sending line {} to child", i);
write_all(stdin, format!("line {}\n", i).into_bytes()).map(|p| p.0)
})
.map(|_| ());
// Try to read `n + 1` lines, ensuring the last one is empty
// (i.e. EOF is reached after `n` lines.
@@ -41,15 +42,15 @@ fn feed_cat(mut cat: Child, n: usize) -> Box<Future<Item = ExitStatus, Error = i
let done = i >= 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);