From d593c5b051f07bde5117122216a356632986b6dd Mon Sep 17 00:00:00 2001 From: Artem Vorotnikov Date: Sat, 14 Dec 2019 09:01:47 +0300 Subject: [PATCH] chore: remove benches and fix/work around clippy lints (#1952) --- examples/chat.rs | 4 +- examples/connect.rs | 9 +- examples/echo-udp.rs | 4 +- examples/echo.rs | 4 +- examples/print_each_packet.rs | 4 +- examples/proxy.rs | 8 +- examples/tinydb.rs | 19 +- examples/tinyhttp.rs | 4 +- examples/udp-client.rs | 2 +- examples/udp-codec.rs | 4 +- tokio-test/src/task.rs | 2 + tokio-test/tests/block_on.rs | 10 +- tokio-util/tests/framed_write.rs | 2 +- tokio/benches/latency.rs | 114 -------- tokio/benches/mio-ops.rs | 57 ---- tokio/benches/mpsc.rs | 270 ------------------- tokio/benches/oneshot.rs | 120 --------- tokio/benches/tcp.rs | 257 ------------------ tokio/benches/thread_pool.rs | 153 ----------- tokio/src/lib.rs | 1 + tokio/src/macros/assert.rs | 5 +- tokio/src/runtime/builder.rs | 2 +- tokio/src/runtime/shell.rs | 2 + tokio/src/runtime/thread_pool/tests/queue.rs | 14 +- tokio/src/signal/registry.rs | 2 + tokio/src/time/tests/test_queue.rs | 44 ++- tokio/tests/io_async_read.rs | 1 + tokio/tests/process_issue_42.rs | 4 +- tokio/tests/rt_common.rs | 1 + tokio/tests/support/mock_file.rs | 2 + tokio/tests/sync_barrier.rs | 1 + tokio/tests/sync_mpsc.rs | 1 + tokio/tests/sync_watch.rs | 1 + tokio/tests/tcp_peek.rs | 2 +- 34 files changed, 89 insertions(+), 1041 deletions(-) delete mode 100644 tokio/benches/latency.rs delete mode 100644 tokio/benches/mio-ops.rs delete mode 100644 tokio/benches/mpsc.rs delete mode 100644 tokio/benches/oneshot.rs delete mode 100644 tokio/benches/tcp.rs delete mode 100644 tokio/benches/thread_pool.rs diff --git a/examples/chat.rs b/examples/chat.rs index 2553cc5ed..91589072b 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -49,7 +49,9 @@ async fn main() -> Result<(), Box> { // client connection. let state = Arc::new(Mutex::new(Shared::new())); - let addr = env::args().nth(1).unwrap_or("127.0.0.1:6142".to_string()); + let addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:6142".to_string()); // Bind a TCP listener to the socket address. // diff --git a/examples/connect.rs b/examples/connect.rs index cdd18e197..d51af88c9 100644 --- a/examples/connect.rs +++ b/examples/connect.rs @@ -36,10 +36,9 @@ async fn main() -> Result<(), Box> { }; // Parse what address we're going to connect to - let addr = match args.first() { - Some(addr) => addr, - None => Err("this program requires at least one argument")?, - }; + let addr = args + .first() + .ok_or("this program requires at least one argument")?; let addr = addr.parse::()?; let stdin = FramedRead::new(io::stdin(), codec::Bytes); @@ -163,7 +162,7 @@ mod codec { type Error = io::Error; fn decode(&mut self, buf: &mut BytesMut) -> io::Result>> { - if buf.len() > 0 { + if !buf.is_empty() { let len = buf.len(); Ok(Some(buf.split_to(len).into_iter().collect())) } else { diff --git a/examples/echo-udp.rs b/examples/echo-udp.rs index f1e8134df..d8b2af9cb 100644 --- a/examples/echo-udp.rs +++ b/examples/echo-udp.rs @@ -51,7 +51,9 @@ impl Server { #[tokio::main] async fn main() -> Result<(), Box> { - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:8080".to_string()); let socket = UdpSocket::bind(&addr).await?; println!("Listening on: {}", socket.local_addr()?); diff --git a/examples/echo.rs b/examples/echo.rs index 455aebde0..35b122794 100644 --- a/examples/echo.rs +++ b/examples/echo.rs @@ -33,7 +33,9 @@ async fn main() -> Result<(), Box> { // Allow passing an address to listen on as the first argument of this // program, but otherwise we'll just set up our TCP listener on // 127.0.0.1:8080 for connections. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:8080".to_string()); // Next up we create a TCP listener which will listen for incoming // connections. This TCP listener is bound to the address we determined diff --git a/examples/print_each_packet.rs b/examples/print_each_packet.rs index f056db4ab..4604139b4 100644 --- a/examples/print_each_packet.rs +++ b/examples/print_each_packet.rs @@ -65,7 +65,9 @@ async fn main() -> Result<(), Box> { // Allow passing an address to listen on as the first argument of this // program, but otherwise we'll just set up our TCP listener on // 127.0.0.1:8080 for connections. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:8080".to_string()); // Next up we create a TCP listener which will listen for incoming // connections. This TCP listener is bound to the address we determined diff --git a/examples/proxy.rs b/examples/proxy.rs index 48f8f0572..f7a9111f6 100644 --- a/examples/proxy.rs +++ b/examples/proxy.rs @@ -32,8 +32,12 @@ use std::error::Error; #[tokio::main] async fn main() -> Result<(), Box> { - let listen_addr = env::args().nth(1).unwrap_or("127.0.0.1:8081".to_string()); - let server_addr = env::args().nth(2).unwrap_or("127.0.0.1:8080".to_string()); + let listen_addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:8081".to_string()); + let server_addr = env::args() + .nth(2) + .unwrap_or_else(|| "127.0.0.1:8080".to_string()); println!("Listening on: {}", listen_addr); println!("Proxying to: {}", server_addr); diff --git a/examples/tinydb.rs b/examples/tinydb.rs index 3fc88f6bd..cf867a0a6 100644 --- a/examples/tinydb.rs +++ b/examples/tinydb.rs @@ -84,7 +84,9 @@ enum Response { async fn main() -> Result<(), Box> { // Parse the address we're going to run this server on // and set up our TCP listener to accept connections. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:8080".to_string()); let mut listener = TcpListener::bind(&addr).await?; println!("Listening on: {}", addr); @@ -175,15 +177,12 @@ fn handle_request(line: &str, db: &Arc) -> Response { impl Request { fn parse(input: &str) -> Result { - let mut parts = input.splitn(3, " "); + let mut parts = input.splitn(3, ' '); match parts.next() { Some("GET") => { - let key = match parts.next() { - Some(key) => key, - None => return Err(format!("GET must be followed by a key")), - }; + let key = parts.next().ok_or("GET must be followed by a key")?; if parts.next().is_some() { - return Err(format!("GET's key must not be followed by anything")); + return Err("GET's key must not be followed by anything".into()); } Ok(Request::Get { key: key.to_string(), @@ -192,11 +191,11 @@ impl Request { Some("SET") => { let key = match parts.next() { Some(key) => key, - None => return Err(format!("SET must be followed by a key")), + None => return Err("SET must be followed by a key".into()), }; let value = match parts.next() { Some(value) => value, - None => return Err(format!("SET needs a value")), + None => return Err("SET needs a value".into()), }; Ok(Request::Set { key: key.to_string(), @@ -204,7 +203,7 @@ impl Request { }) } Some(cmd) => Err(format!("unknown command: {}", cmd)), - None => Err(format!("empty input")), + None => Err("empty input".into()), } } } diff --git a/examples/tinyhttp.rs b/examples/tinyhttp.rs index f8731b9fc..5ddf0d486 100644 --- a/examples/tinyhttp.rs +++ b/examples/tinyhttp.rs @@ -27,7 +27,9 @@ use tokio_util::codec::{Decoder, Encoder, Framed}; async fn main() -> Result<(), Box> { // Parse the arguments, bind the TCP socket we'll be listening to, spin up // our worker threads, and start shipping sockets to those worker threads. - let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:8080".to_string()); let mut server = TcpListener::bind(&addr).await?; let mut incoming = server.incoming(); println!("Listening on: {}", addr); diff --git a/examples/udp-client.rs b/examples/udp-client.rs index 5437daf60..a191033d8 100644 --- a/examples/udp-client.rs +++ b/examples/udp-client.rs @@ -44,7 +44,7 @@ fn get_stdin_data() -> Result, Box> { async fn main() -> Result<(), Box> { let remote_addr: SocketAddr = env::args() .nth(1) - .unwrap_or("127.0.0.1:8080".into()) + .unwrap_or_else(|| "127.0.0.1:8080".into()) .parse()?; // We use port 0 to let the operating system allocate an available port for us. diff --git a/examples/udp-codec.rs b/examples/udp-codec.rs index 0c9dbf76e..6b3f84a0f 100644 --- a/examples/udp-codec.rs +++ b/examples/udp-codec.rs @@ -22,7 +22,9 @@ use std::time::Duration; #[tokio::main] async fn main() -> Result<(), Box> { - let addr = env::args().nth(1).unwrap_or("127.0.0.1:0".to_string()); + let addr = env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:0".to_string()); // Bind both our sockets and then figure out what ports we got. let a = UdpSocket::bind(&addr).await?; diff --git a/tokio-test/src/task.rs b/tokio-test/src/task.rs index c21e31a5a..4790de542 100644 --- a/tokio-test/src/task.rs +++ b/tokio-test/src/task.rs @@ -1,5 +1,7 @@ //! Futures task based helpers +#![allow(clippy::mutex_atomic)] + use futures_core::Stream; use std::future::Future; use std::mem; diff --git a/tokio-test/tests/block_on.rs b/tokio-test/tests/block_on.rs index 7aec82ccd..d640a13c9 100644 --- a/tokio-test/tests/block_on.rs +++ b/tokio-test/tests/block_on.rs @@ -20,10 +20,8 @@ fn async_fn() { #[test] fn test_delay() { let deadline = Instant::now() + Duration::from_millis(100); - assert_eq!( - (), - block_on(async { - delay_until(deadline).await; - }) - ); + + block_on(async { + delay_until(deadline).await; + }); } diff --git a/tokio-util/tests/framed_write.rs b/tokio-util/tests/framed_write.rs index 706e6792f..b2970c39d 100644 --- a/tokio-util/tests/framed_write.rs +++ b/tokio-util/tests/framed_write.rs @@ -82,7 +82,7 @@ fn write_hits_backpressure() { // Append to the end match mock.calls.back_mut().unwrap() { - &mut Ok(ref mut data) => { + Ok(ref mut data) => { // Write in 2kb chunks if data.len() < ITER { data.extend_from_slice(&b[..]); diff --git a/tokio/benches/latency.rs b/tokio/benches/latency.rs deleted file mode 100644 index b44335fde..000000000 --- a/tokio/benches/latency.rs +++ /dev/null @@ -1,114 +0,0 @@ -#![cfg(feature = "broken")] -#![feature(test)] -#![warn(rust_2018_idioms)] - -extern crate test; - -use std::io; -use std::net::SocketAddr; -use std::thread; - -use futures::sync::mpsc; -use futures::sync::oneshot; -use futures::try_ready; -use futures::{Future, Poll, Sink, Stream}; -use test::Bencher; -use tokio::net::UdpSocket; - -/// UDP echo server -struct EchoServer { - socket: UdpSocket, - buf: Vec, - to_send: Option<(usize, SocketAddr)>, -} - -impl EchoServer { - fn new(s: UdpSocket) -> Self { - EchoServer { - socket: s, - to_send: None, - buf: vec![0u8; 1600], - } - } -} - -impl Future for EchoServer { - type Item = (); - type Error = io::Error; - - fn poll(&mut self) -> Poll<(), io::Error> { - loop { - if let Some(&(size, peer)) = self.to_send.as_ref() { - try_ready!(self.socket.poll_send_to(&self.buf[..size], &peer)); - self.to_send = None; - } - self.to_send = Some(try_ready!(self.socket.poll_recv_from(&mut self.buf))); - } - } -} - -#[bench] -fn udp_echo_latency(b: &mut Bencher) { - let any_addr = "127.0.0.1:0".to_string(); - let any_addr = any_addr.parse::().unwrap(); - - let (stop_c, stop_p) = oneshot::channel::<()>(); - let (tx, rx) = oneshot::channel(); - - let child = thread::spawn(move || { - let socket = tokio::net::UdpSocket::bind(&any_addr).unwrap(); - tx.send(socket.local_addr().unwrap()).unwrap(); - - let server = EchoServer::new(socket); - let server = server.select(stop_p.map_err(|_| panic!())); - let server = server.map_err(|_| ()); - server.wait().unwrap(); - }); - - let client = std::net::UdpSocket::bind(&any_addr).unwrap(); - - let server_addr = rx.wait().unwrap(); - let mut buf = [0u8; 1000]; - - // warmup phase; for some reason initial couple of - // runs are much slower - // - // TODO: Describe the exact reasons; caching? branch predictor? lazy closures? - for _ in 0..8 { - client.send_to(&buf, &server_addr).unwrap(); - let _ = client.recv_from(&mut buf).unwrap(); - } - - b.iter(|| { - client.send_to(&buf, &server_addr).unwrap(); - let _ = client.recv_from(&mut buf).unwrap(); - }); - - stop_c.send(()).unwrap(); - child.join().unwrap(); -} - -#[bench] -fn futures_channel_latency(b: &mut Bencher) { - let (mut in_tx, in_rx) = mpsc::channel(32); - let (out_tx, out_rx) = mpsc::channel::<_>(32); - - let child = thread::spawn(|| out_tx.send_all(in_rx.then(|r| r.unwrap())).wait()); - let mut rx_iter = out_rx.wait(); - - // warmup phase; for some reason initial couple of runs are much slower - // - // TODO: Describe the exact reasons; caching? branch predictor? lazy closures? - for _ in 0..8 { - in_tx.start_send(Ok(1usize)).unwrap(); - let _ = rx_iter.next(); - } - - b.iter(|| { - in_tx.start_send(Ok(1usize)).unwrap(); - let _ = rx_iter.next(); - }); - - drop(in_tx); - child.join().unwrap().unwrap(); -} diff --git a/tokio/benches/mio-ops.rs b/tokio/benches/mio-ops.rs deleted file mode 100644 index 8aedbfd8e..000000000 --- a/tokio/benches/mio-ops.rs +++ /dev/null @@ -1,57 +0,0 @@ -// Measure cost of different operations -// to get a sense of performance tradeoffs -#![cfg(feature = "broken")] -#![feature(test)] -#![warn(rust_2018_idioms)] - -extern crate test; - -use test::Bencher; - -use mio::tcp::TcpListener; -use mio::{PollOpt, Ready, Token}; - -#[bench] -fn mio_register_deregister(b: &mut Bencher) { - let addr = "127.0.0.1:0".parse().unwrap(); - // Setup the server socket - let sock = TcpListener::bind(&addr).unwrap(); - let poll = mio::Poll::new().unwrap(); - - const CLIENT: Token = Token(1); - - b.iter(|| { - poll.register(&sock, CLIENT, Ready::readable(), PollOpt::edge()) - .unwrap(); - poll.deregister(&sock).unwrap(); - }); -} - -#[bench] -fn mio_reregister(b: &mut Bencher) { - let addr = "127.0.0.1:0".parse().unwrap(); - // Setup the server socket - let sock = TcpListener::bind(&addr).unwrap(); - let poll = mio::Poll::new().unwrap(); - - const CLIENT: Token = Token(1); - poll.register(&sock, CLIENT, Ready::readable(), PollOpt::edge()) - .unwrap(); - - b.iter(|| { - poll.reregister(&sock, CLIENT, Ready::readable(), PollOpt::edge()) - .unwrap(); - }); - poll.deregister(&sock).unwrap(); -} - -#[bench] -fn mio_poll(b: &mut Bencher) { - let poll = mio::Poll::new().unwrap(); - let timeout = std::time::Duration::new(0, 0); - let mut events = mio::Events::with_capacity(1024); - - b.iter(|| { - poll.poll(&mut events, Some(timeout)).unwrap(); - }); -} diff --git a/tokio/benches/mpsc.rs b/tokio/benches/mpsc.rs deleted file mode 100644 index 0b97d55d3..000000000 --- a/tokio/benches/mpsc.rs +++ /dev/null @@ -1,270 +0,0 @@ -#![feature(test)] -#![warn(rust_2018_idioms)] - -extern crate test; - -use tokio::sync::mpsc::*; - -use futures::{future, Async, Future, Sink, Stream}; -use std::thread; -use test::Bencher; - -type Medium = [usize; 64]; -type Large = [Medium; 64]; - -#[bench] -fn bounded_new_medium(b: &mut Bencher) { - b.iter(|| { - let _ = test::black_box(&channel::(1_000)); - }) -} - -#[bench] -fn unbounded_new_medium(b: &mut Bencher) { - b.iter(|| { - let _ = test::black_box(&unbounded_channel::()); - }) -} -#[bench] -fn bounded_new_large(b: &mut Bencher) { - b.iter(|| { - let _ = test::black_box(&channel::(1_000)); - }) -} - -#[bench] -fn unbounded_new_large(b: &mut Bencher) { - b.iter(|| { - let _ = test::black_box(&unbounded_channel::()); - }) -} - -#[bench] -fn send_one_message(b: &mut Bencher) { - b.iter(|| { - let (mut tx, mut rx) = channel(1_000); - - // Send - tx.try_send(1).unwrap(); - - // Receive - assert_eq!(Async::Ready(Some(1)), rx.poll().unwrap()); - }) -} - -#[bench] -fn send_one_message_large(b: &mut Bencher) { - b.iter(|| { - let (mut tx, mut rx) = channel::(1_000); - - // Send - let _ = tx.try_send([[0; 64]; 64]); - - // Receive - let _ = test::black_box(&rx.poll()); - }) -} - -#[bench] -fn bounded_rx_not_ready(b: &mut Bencher) { - let (_tx, mut rx) = channel::(1_000); - b.iter(|| { - future::lazy(|| { - assert!(rx.poll().unwrap().is_not_ready()); - - Ok::<_, ()>(()) - }) - .wait() - .unwrap(); - }) -} - -#[bench] -fn bounded_tx_poll_ready(b: &mut Bencher) { - let (mut tx, _rx) = channel::(1); - b.iter(|| { - future::lazy(|| { - assert!(tx.poll_ready().unwrap().is_ready()); - - Ok::<_, ()>(()) - }) - .wait() - .unwrap(); - }) -} - -#[bench] -fn bounded_tx_poll_not_ready(b: &mut Bencher) { - let (mut tx, _rx) = channel::(1); - tx.try_send(1).unwrap(); - b.iter(|| { - future::lazy(|| { - assert!(tx.poll_ready().unwrap().is_not_ready()); - - Ok::<_, ()>(()) - }) - .wait() - .unwrap(); - }) -} - -#[bench] -fn unbounded_rx_not_ready(b: &mut Bencher) { - let (_tx, mut rx) = unbounded_channel::(); - b.iter(|| { - future::lazy(|| { - assert!(rx.poll().unwrap().is_not_ready()); - - Ok::<_, ()>(()) - }) - .wait() - .unwrap(); - }) -} - -#[bench] -fn unbounded_rx_not_ready_x5(b: &mut Bencher) { - let (_tx, mut rx) = unbounded_channel::(); - b.iter(|| { - future::lazy(|| { - assert!(rx.poll().unwrap().is_not_ready()); - assert!(rx.poll().unwrap().is_not_ready()); - assert!(rx.poll().unwrap().is_not_ready()); - assert!(rx.poll().unwrap().is_not_ready()); - assert!(rx.poll().unwrap().is_not_ready()); - - Ok::<_, ()>(()) - }) - .wait() - .unwrap(); - }) -} - -#[bench] -fn bounded_uncontended_1(b: &mut Bencher) { - b.iter(|| { - let (mut tx, mut rx) = channel(1_000); - - for i in 0..1000 { - tx.try_send(i).unwrap(); - // No need to create a task, because poll is not going to park. - assert_eq!(Async::Ready(Some(i)), rx.poll().unwrap()); - } - }) -} - -#[bench] -fn bounded_uncontended_1_large(b: &mut Bencher) { - b.iter(|| { - let (mut tx, mut rx) = channel::(1_000); - - for i in 0..1000 { - let _ = tx.try_send([[i; 64]; 64]); - // No need to create a task, because poll is not going to park. - let _ = test::black_box(&rx.poll()); - } - }) -} - -#[bench] -fn bounded_uncontended_2(b: &mut Bencher) { - b.iter(|| { - let (mut tx, mut rx) = channel(1000); - - for i in 0..1000 { - tx.try_send(i).unwrap(); - } - - for i in 0..1000 { - // No need to create a task, because poll is not going to park. - assert_eq!(Async::Ready(Some(i)), rx.poll().unwrap()); - } - }) -} - -#[bench] -fn contended_unbounded_tx(b: &mut Bencher) { - let mut threads = vec![]; - let mut txs = vec![]; - - for _ in 0..4 { - let (tx, rx) = ::std::sync::mpsc::channel::>(); - txs.push(tx); - - threads.push(thread::spawn(move || { - for mut tx in rx.iter() { - for i in 0..1_000 { - tx.try_send(i).unwrap(); - } - } - })); - } - - b.iter(|| { - // TODO make unbounded - let (tx, rx) = channel::(1_000_000); - - for th in &txs { - th.send(tx.clone()).unwrap(); - } - - drop(tx); - - let rx = rx.wait().take(4 * 1_000); - - for v in rx { - let _ = test::black_box(v); - } - }); - - drop(txs); - - for th in threads { - th.join().unwrap(); - } -} - -#[bench] -fn contended_bounded_tx(b: &mut Bencher) { - const THREADS: usize = 4; - const ITERS: usize = 100; - - let mut threads = vec![]; - let mut txs = vec![]; - - for _ in 0..THREADS { - let (tx, rx) = ::std::sync::mpsc::channel::>(); - txs.push(tx); - - threads.push(thread::spawn(move || { - for tx in rx.iter() { - let mut tx = tx.wait(); - for i in 0..ITERS { - tx.send(i as i32).unwrap(); - } - } - })); - } - - b.iter(|| { - let (tx, rx) = channel::(1); - - for th in &txs { - th.send(tx.clone()).unwrap(); - } - - drop(tx); - - let rx = rx.wait().take(THREADS * ITERS); - - for v in rx { - let _ = test::black_box(v); - } - }); - - drop(txs); - - for th in threads { - th.join().unwrap(); - } -} diff --git a/tokio/benches/oneshot.rs b/tokio/benches/oneshot.rs deleted file mode 100644 index a7f43c2f6..000000000 --- a/tokio/benches/oneshot.rs +++ /dev/null @@ -1,120 +0,0 @@ -#![feature(test)] -#![warn(rust_2018_idioms)] - -extern crate test; - -use tokio::sync::oneshot; - -use futures::{future, Async, Future}; -use test::Bencher; - -#[bench] -fn new(b: &mut Bencher) { - b.iter(|| { - let _ = ::test::black_box(&oneshot::channel::()); - }) -} - -#[bench] -fn same_thread_send_recv(b: &mut Bencher) { - b.iter(|| { - let (tx, mut rx) = oneshot::channel(); - - let _ = tx.send(1); - - assert_eq!(Async::Ready(1), rx.poll().unwrap()); - }); -} - -#[bench] -fn same_thread_recv_multi_send_recv(b: &mut Bencher) { - b.iter(|| { - let (tx, mut rx) = oneshot::channel(); - - future::lazy(|| { - let _ = rx.poll(); - let _ = rx.poll(); - let _ = rx.poll(); - let _ = rx.poll(); - - let _ = tx.send(1); - assert_eq!(Async::Ready(1), rx.poll().unwrap()); - - Ok::<_, ()>(()) - }) - .wait() - .unwrap(); - }); -} - -#[bench] -fn multi_thread_send_recv(b: &mut Bencher) { - const MAX: usize = 10_000_000; - - use std::thread; - - fn spin(mut f: F) -> Result { - use futures::Async::Ready; - loop { - match f.poll() { - Ok(Ready(v)) => return Ok(v), - Ok(_) => {} - Err(e) => return Err(e), - } - } - } - - let mut ping_txs = vec![]; - let mut ping_rxs = vec![]; - let mut pong_txs = vec![]; - let mut pong_rxs = vec![]; - - for _ in 0..MAX { - let (tx, rx) = oneshot::channel::<()>(); - - ping_txs.push(Some(tx)); - ping_rxs.push(Some(rx)); - - let (tx, rx) = oneshot::channel::<()>(); - - pong_txs.push(Some(tx)); - pong_rxs.push(Some(rx)); - } - - thread::spawn(move || { - future::lazy(|| { - for i in 0..MAX { - let ping_rx = ping_rxs[i].take().unwrap(); - let pong_tx = pong_txs[i].take().unwrap(); - - if spin(ping_rx).is_err() { - return Ok(()); - } - - pong_tx.send(()).unwrap(); - } - - Ok::<(), ()>(()) - }) - .wait() - .unwrap(); - }); - - future::lazy(|| { - let mut i = 0; - - b.iter(|| { - let ping_tx = ping_txs[i].take().unwrap(); - let pong_rx = pong_rxs[i].take().unwrap(); - - ping_tx.send(()).unwrap(); - spin(pong_rx).unwrap(); - - i += 1; - }); - - Ok::<(), ()>(()) - }) - .wait() - .unwrap(); -} diff --git a/tokio/benches/tcp.rs b/tokio/benches/tcp.rs deleted file mode 100644 index f9a4a03b1..000000000 --- a/tokio/benches/tcp.rs +++ /dev/null @@ -1,257 +0,0 @@ -#![cfg(feature = "broken")] -#![feature(test)] -#![warn(rust_2018_idioms)] - -pub extern crate test; - -mod prelude { - pub use futures::*; - pub use tokio::net::{TcpListener, TcpStream}; - pub use tokio::reactor::Reactor; - pub use tokio_io::io::read_to_end; - - pub use std::io::{self, Read, Write}; - pub use std::thread; - pub use std::time::Duration; - pub use test::{self, Bencher}; -} - -mod connect_churn { - use crate::prelude::*; - - const NUM: usize = 300; - const CONCURRENT: usize = 8; - - #[bench] - fn one_thread(b: &mut Bencher) { - let addr = "127.0.0.1:0".parse().unwrap(); - - b.iter(move || { - let listener = TcpListener::bind(&addr).unwrap(); - let addr = listener.local_addr().unwrap(); - - // Spawn a single future that accepts & drops connections - let serve_incomings = listener - .incoming() - .map_err(|e| panic!("server err: {:?}", e)) - .for_each(|_| Ok(())); - - let connects = stream::iter_result((0..NUM).map(|_| { - Ok(TcpStream::connect(&addr).and_then(|sock| { - sock.set_linger(Some(Duration::from_secs(0))).unwrap(); - read_to_end(sock, vec![]) - })) - })); - - let connects_concurrent = connects - .buffer_unordered(CONCURRENT) - .map_err(|e| panic!("client err: {:?}", e)) - .for_each(|_| Ok(())); - - serve_incomings - .select(connects_concurrent) - .map(|_| ()) - .map_err(|_| ()) - .wait() - .unwrap(); - }); - } - - fn n_workers(n: usize, b: &mut Bencher) { - let (shutdown_tx, shutdown_rx) = sync::oneshot::channel(); - let (addr_tx, addr_rx) = sync::oneshot::channel(); - - // Spawn reactor thread - let server_thread = thread::spawn(move || { - // Bind the TCP listener - let listener = TcpListener::bind(&"127.0.0.1:0".parse().unwrap()).unwrap(); - - // Get the address being listened on. - let addr = listener.local_addr().unwrap(); - - // Send the remote & address back to the main thread - addr_tx.send(addr).unwrap(); - - // Spawn a single future that accepts & drops connections - let serve_incomings = listener - .incoming() - .map_err(|e| panic!("server err: {:?}", e)) - .for_each(|_| Ok(())); - - // Run server - serve_incomings - .select(shutdown_rx) - .map(|_| ()) - .map_err(|_| ()) - .wait() - .unwrap(); - }); - - // Get the bind addr of the server - let addr = addr_rx.wait().unwrap(); - - b.iter(move || { - use std::sync::{Arc, Barrier}; - - // Create a barrier to coordinate threads - let barrier = Arc::new(Barrier::new(n + 1)); - - // Spawn worker threads - let threads: Vec<_> = (0..n) - .map(|_| { - let barrier = barrier.clone(); - let addr = addr.clone(); - - thread::spawn(move || { - let connects = stream::iter_result((0..(NUM / n)).map(|_| { - Ok(TcpStream::connect(&addr) - .map_err(|e| panic!("connect err: {:?}", e)) - .and_then(|sock| { - sock.set_linger(Some(Duration::from_secs(0))).unwrap(); - read_to_end(sock, vec![]) - })) - })); - - barrier.wait(); - - connects - .buffer_unordered(CONCURRENT) - .map_err(|e| panic!("client err: {:?}", e)) - .for_each(|_| Ok(())) - .wait() - .unwrap(); - }) - }) - .collect(); - - barrier.wait(); - - for th in threads { - th.join().unwrap(); - } - }); - - // Shutdown the server - shutdown_tx.send(()).unwrap(); - server_thread.join().unwrap(); - } - - #[bench] - fn two_threads(b: &mut Bencher) { - n_workers(1, b); - } - - #[bench] - fn multi_threads(b: &mut Bencher) { - n_workers(4, b); - } -} - -mod transfer { - use crate::prelude::*; - use std::{cmp, mem}; - use tokio_io::try_nb; - - const MB: usize = 3 * 1024 * 1024; - - struct Drain { - sock: TcpStream, - chunk: usize, - } - - impl Future for Drain { - type Item = (); - type Error = io::Error; - - fn poll(&mut self) -> Poll<(), io::Error> { - let mut buf: [u8; 1024] = unsafe { mem::uninitialized() }; - - loop { - match try_nb!(self.sock.read(&mut buf[..self.chunk])) { - 0 => return Ok(Async::Ready(())), - _ => {} - } - } - } - } - - struct Transfer { - sock: TcpStream, - rem: usize, - chunk: usize, - } - - impl Future for Transfer { - type Item = (); - type Error = io::Error; - - fn poll(&mut self) -> Poll<(), io::Error> { - while self.rem > 0 { - let len = cmp::min(self.rem, self.chunk); - let buf = &DATA[..len]; - - let n = try_nb!(self.sock.write(&buf)); - self.rem -= n; - } - - Ok(Async::Ready(())) - } - } - - static DATA: [u8; 1024] = [0; 1024]; - - fn one_thread(b: &mut Bencher, read_size: usize, write_size: usize) { - let addr = "127.0.0.1:0".parse().unwrap(); - - b.iter(move || { - let listener = TcpListener::bind(&addr).unwrap(); - let addr = listener.local_addr().unwrap(); - - // Spawn a single future that accepts 1 connection, Drain it and drops - let server = listener - .incoming() - .into_future() // take the first connection - .map_err(|(e, _other_incomings)| e) - .map(|(connection, _other_incomings)| connection.unwrap()) - .and_then(|sock| { - sock.set_linger(Some(Duration::from_secs(0))).unwrap(); - let drain = Drain { - sock, - chunk: read_size, - }; - drain - .map(|_| ()) - .map_err(|e| panic!("server error: {:?}", e)) - }) - .map_err(|e| panic!("server err: {:?}", e)); - - let client = TcpStream::connect(&addr) - .and_then(move |sock| Transfer { - sock, - rem: MB, - chunk: write_size, - }) - .map_err(|e| panic!("client err: {:?}", e)); - - server.join(client).wait().unwrap(); - }); - } - - mod small_chunks { - use crate::prelude::*; - - #[bench] - fn one_thread(b: &mut Bencher) { - super::one_thread(b, 32, 32); - } - } - - mod big_chunks { - use crate::prelude::*; - - #[bench] - fn one_thread(b: &mut Bencher) { - super::one_thread(b, 1_024, 1_024); - } - } -} diff --git a/tokio/benches/thread_pool.rs b/tokio/benches/thread_pool.rs deleted file mode 100644 index 90aaf06f8..000000000 --- a/tokio/benches/thread_pool.rs +++ /dev/null @@ -1,153 +0,0 @@ -#![feature(test)] - -extern crate test; - -use tokio::runtime::Builder; -use tokio::sync::oneshot; - -use std::future::Future; -use std::pin::Pin; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering::Relaxed; -use std::sync::{mpsc, Arc}; -use std::task::{Context, Poll}; - -struct Backoff(usize); - -impl Future for Backoff { - type Output = (); - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { - if self.0 == 0 { - Poll::Ready(()) - } else { - self.0 -= 1; - cx.waker().wake_by_ref(); - Poll::Pending - } - } -} - -#[bench] -fn spawn_many(b: &mut test::Bencher) { - const NUM_SPAWN: usize = 10_000; - - let rt = Builder::new().threaded_scheduler().build().unwrap(); - - let (tx, rx) = mpsc::sync_channel(1000); - let rem = Arc::new(AtomicUsize::new(0)); - - b.iter(|| { - rem.store(NUM_SPAWN, Relaxed); - - for _ in 0..NUM_SPAWN { - let tx = tx.clone(); - let rem = rem.clone(); - - rt.spawn(async move { - if 1 == rem.fetch_sub(1, Relaxed) { - tx.send(()).unwrap(); - } - }); - } - - let _ = rx.recv().unwrap(); - }); -} - -#[bench] -fn yield_many(b: &mut test::Bencher) { - const NUM_YIELD: usize = 1_000; - const TASKS_PER_CPU: usize = 50; - - let rt = Builder::new().threaded_scheduler().build().unwrap(); - - let tasks = TASKS_PER_CPU * num_cpus::get_physical(); - let (tx, rx) = mpsc::sync_channel(tasks); - - b.iter(move || { - for _ in 0..tasks { - let tx = tx.clone(); - - rt.spawn(async move { - let backoff = Backoff(NUM_YIELD); - backoff.await; - tx.send(()).unwrap(); - }); - } - - for _ in 0..tasks { - let _ = rx.recv().unwrap(); - } - }); -} - -#[bench] -fn ping_pong(b: &mut test::Bencher) { - const NUM_PINGS: usize = 1_000; - - let rt = Builder::new().threaded_scheduler().build().unwrap(); - - let (done_tx, done_rx) = mpsc::sync_channel(1000); - let rem = Arc::new(AtomicUsize::new(0)); - - b.iter(|| { - let done_tx = done_tx.clone(); - let rem = rem.clone(); - rem.store(NUM_PINGS, Relaxed); - - rt.spawn(async move { - for _ in 0..NUM_PINGS { - let rem = rem.clone(); - let done_tx = done_tx.clone(); - - tokio::spawn(async move { - let (tx1, rx1) = oneshot::channel(); - let (tx2, rx2) = oneshot::channel(); - - tokio::spawn(async move { - rx1.await.unwrap(); - tx2.send(()).unwrap(); - }); - - tx1.send(()).unwrap(); - rx2.await.unwrap(); - - if 1 == rem.fetch_sub(1, Relaxed) { - done_tx.send(()).unwrap(); - } - }); - } - }); - - done_rx.recv().unwrap(); - }); -} - -#[bench] -fn chained_spawn(b: &mut test::Bencher) { - const ITER: usize = 1_000; - - let rt = Builder::new().threaded_scheduler().build().unwrap(); - - fn iter(done_tx: mpsc::SyncSender<()>, n: usize) { - if n == 0 { - done_tx.send(()).unwrap(); - } else { - tokio::spawn(async move { - iter(done_tx, n - 1); - }); - } - } - - let (done_tx, done_rx) = mpsc::sync_channel(1000); - - b.iter(move || { - let done_tx = done_tx.clone(); - rt.spawn(async move { - iter(done_tx, ITER); - }); - - done_rx.recv().unwrap(); - }); -} diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index 5cbddf66f..e617ec015 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -1,4 +1,5 @@ #![doc(html_root_url = "https://docs.rs/tokio/0.2.4")] +#![allow(clippy::cognitive_complexity)] #![warn( missing_debug_implementations, missing_docs, diff --git a/tokio/src/macros/assert.rs b/tokio/src/macros/assert.rs index fd6601b44..4f5760921 100644 --- a/tokio/src/macros/assert.rs +++ b/tokio/src/macros/assert.rs @@ -11,9 +11,8 @@ macro_rules! assert_some { /// Assert option is none macro_rules! assert_none { ($e:expr) => {{ - match $e { - Some(v) => panic!("expected none, was {:?}", v), - _ => {} + if let Some(v) = $e { + panic!("expected none, was {:?}", v); } }}; } diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index 1b51379b4..802608e53 100644 --- a/tokio/src/runtime/builder.rs +++ b/tokio/src/runtime/builder.rs @@ -430,7 +430,7 @@ cfg_rt_threaded! { io_handle, time_handle, clock, - blocking_spawner: blocking_spawner.clone(), + blocking_spawner, }, blocking_pool, }) diff --git a/tokio/src/runtime/shell.rs b/tokio/src/runtime/shell.rs index 98d0ee6fe..992244dea 100644 --- a/tokio/src/runtime/shell.rs +++ b/tokio/src/runtime/shell.rs @@ -1,3 +1,5 @@ +#![allow(clippy::redundant_clone)] + use crate::park::Park; use crate::runtime::enter; use crate::runtime::time; diff --git a/tokio/src/runtime/thread_pool/tests/queue.rs b/tokio/src/runtime/thread_pool/tests/queue.rs index ef0e373c7..83a602758 100644 --- a/tokio/src/runtime/thread_pool/tests/queue.rs +++ b/tokio/src/runtime/thread_pool/tests/queue.rs @@ -41,9 +41,8 @@ macro_rules! assert_steal { macro_rules! assert_empty { ($q:expr) => {{ let q: &mut queue::Worker = &mut $q; - match q.pop_local_first() { - Some(v) => panic!("expected emtpy queue; got {}", num(v)), - None => {} + if let Some(v) = q.pop_local_first() { + panic!("expected emtpy queue; got {}", num(v)); } }}; } @@ -261,12 +260,9 @@ fn num(task: Task) -> u32 { for (_, join) in map.iter_mut() { let mut cx = Context::from_waker(noop_waker_ref()); - match Pin::new(join).poll(&mut cx) { - Ready(n) => { - num = Some(n.unwrap()); - break; - } - _ => {} + if let Ready(n) = Pin::new(join).poll(&mut cx) { + num = Some(n.unwrap()); + break; } } diff --git a/tokio/src/signal/registry.rs b/tokio/src/signal/registry.rs index e4bfe758f..d5b44cce6 100644 --- a/tokio/src/signal/registry.rs +++ b/tokio/src/signal/registry.rs @@ -1,3 +1,5 @@ +#![allow(clippy::unit_arg)] + use crate::signal::os::{OsExtraData, OsStorage}; use crate::sync::mpsc::Sender; diff --git a/tokio/src/time/tests/test_queue.rs b/tokio/src/time/tests/test_queue.rs index 0902ec4a4..34b9b7da4 100644 --- a/tokio/src/time/tests/test_queue.rs +++ b/tokio/src/time/tests/test_queue.rs @@ -108,13 +108,11 @@ fn multi_delay_at_start() { assert!(queue.is_woken()); assert_ready!(poll!(queue)); assert_pending!(poll!(queue)); - } else { - if queue.is_woken() { - let cascade = &[192, 960]; - assert!(cascade.contains(&elapsed), "elapsed={}", elapsed); + } else if queue.is_woken() { + let cascade = &[192, 960]; + assert!(cascade.contains(&elapsed), "elapsed={}", elapsed); - assert_pending!(poll!(queue)); - } + assert_pending!(poll!(queue)); } } }); @@ -288,14 +286,14 @@ fn multi_reset() { let epoch = clock.now(); - let foo = queue.insert_at("foo", epoch + ms(200)); - let bar = queue.insert_at("bar", epoch + ms(250)); + let one = queue.insert_at("one", epoch + ms(200)); + let two = queue.insert_at("two", epoch + ms(250)); assert_pending!(poll!(queue)); - queue.reset_at(&foo, epoch + ms(300)); - queue.reset_at(&bar, epoch + ms(350)); - queue.reset_at(&foo, epoch + ms(400)); + queue.reset_at(&one, epoch + ms(300)); + queue.reset_at(&two, epoch + ms(350)); + queue.reset_at(&one, epoch + ms(400)); }) } @@ -306,19 +304,19 @@ fn expire_first_key_when_reset_to_expire_earlier() { let epoch = clock.now(); - let foo = queue.insert_at("foo", epoch + ms(200)); - queue.insert_at("bar", epoch + ms(250)); + let one = queue.insert_at("one", epoch + ms(200)); + queue.insert_at("two", epoch + ms(250)); assert_pending!(poll!(queue)); - queue.reset_at(&foo, epoch + ms(100)); + queue.reset_at(&one, epoch + ms(100)); clock.advance(ms(100)); assert!(queue.is_woken()); let entry = assert_ready_ok!(poll!(queue)).into_inner(); - assert_eq!(entry, "foo"); + assert_eq!(entry, "one"); }) } @@ -329,18 +327,18 @@ fn expire_second_key_when_reset_to_expire_earlier() { let epoch = clock.now(); - queue.insert_at("foo", epoch + ms(200)); - let bar = queue.insert_at("bar", epoch + ms(250)); + queue.insert_at("one", epoch + ms(200)); + let two = queue.insert_at("two", epoch + ms(250)); assert_pending!(poll!(queue)); - queue.reset_at(&bar, epoch + ms(100)); + queue.reset_at(&two, epoch + ms(100)); clock.advance(ms(100)); assert!(queue.is_woken()); let entry = assert_ready_ok!(poll!(queue)).into_inner(); - assert_eq!(entry, "bar"); + assert_eq!(entry, "two"); }) } @@ -351,18 +349,18 @@ fn reset_first_expiring_item_to_expire_later() { let epoch = clock.now(); - let foo = queue.insert_at("foo", epoch + ms(200)); - let _bar = queue.insert_at("bar", epoch + ms(250)); + let one = queue.insert_at("one", epoch + ms(200)); + let _two = queue.insert_at("two", epoch + ms(250)); assert_pending!(poll!(queue)); - queue.reset_at(&foo, epoch + ms(300)); + queue.reset_at(&one, epoch + ms(300)); clock.advance(ms(250)); assert!(queue.is_woken()); let entry = assert_ready_ok!(poll!(queue)).into_inner(); - assert_eq!(entry, "bar"); + assert_eq!(entry, "two"); }) } diff --git a/tokio/tests/io_async_read.rs b/tokio/tests/io_async_read.rs index 2be2aa1a1..20440bbde 100644 --- a/tokio/tests/io_async_read.rs +++ b/tokio/tests/io_async_read.rs @@ -1,3 +1,4 @@ +#![allow(clippy::transmute_ptr_to_ptr)] #![warn(rust_2018_idioms)] #![cfg(feature = "full")] diff --git a/tokio/tests/process_issue_42.rs b/tokio/tests/process_issue_42.rs index 022a109a6..aa70af3b5 100644 --- a/tokio/tests/process_issue_42.rs +++ b/tokio/tests/process_issue_42.rs @@ -15,9 +15,9 @@ async fn issue_42() { // We then do this many times (in parallel) in an effort to stress test the // implementation to ensure there are no race conditions. // See alexcrichton/tokio-process#42 for background - let join_handles = (0..10usize).into_iter().map(|_| { + let join_handles = (0..10usize).map(|_| { task::spawn(async { - let processes = (0..10usize).into_iter().map(|i| { + let processes = (0..10usize).map(|i| { Command::new("echo") .arg(format!("I am spawned process #{}", i)) .stdin(Stdio::null()) diff --git a/tokio/tests/rt_common.rs b/tokio/tests/rt_common.rs index 06c966bfe..abb0b6788 100644 --- a/tokio/tests/rt_common.rs +++ b/tokio/tests/rt_common.rs @@ -1,3 +1,4 @@ +#![allow(clippy::needless_range_loop)] #![warn(rust_2018_idioms)] #![cfg(feature = "full")] diff --git a/tokio/tests/support/mock_file.rs b/tokio/tests/support/mock_file.rs index 44aa7b3f7..9895f835e 100644 --- a/tokio/tests/support/mock_file.rs +++ b/tokio/tests/support/mock_file.rs @@ -1,3 +1,5 @@ +#![allow(clippy::unnecessary_operation)] + use std::collections::VecDeque; use std::fmt; use std::fs::{Metadata, Permissions}; diff --git a/tokio/tests/sync_barrier.rs b/tokio/tests/sync_barrier.rs index e6f32c7a5..f280fe860 100644 --- a/tokio/tests/sync_barrier.rs +++ b/tokio/tests/sync_barrier.rs @@ -1,3 +1,4 @@ +#![allow(clippy::unnecessary_operation)] #![warn(rust_2018_idioms)] #![cfg(feature = "full")] diff --git a/tokio/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs index f906d3570..e1f99595d 100644 --- a/tokio/tests/sync_mpsc.rs +++ b/tokio/tests/sync_mpsc.rs @@ -1,3 +1,4 @@ +#![allow(clippy::redundant_clone)] #![warn(rust_2018_idioms)] #![cfg(feature = "full")] diff --git a/tokio/tests/sync_watch.rs b/tokio/tests/sync_watch.rs index f13b145da..409615d45 100644 --- a/tokio/tests/sync_watch.rs +++ b/tokio/tests/sync_watch.rs @@ -1,3 +1,4 @@ +#![allow(clippy::cognitive_complexity)] #![warn(rust_2018_idioms)] #![cfg(feature = "full")] diff --git a/tokio/tests/tcp_peek.rs b/tokio/tests/tcp_peek.rs index c5781daf2..aecc0ac19 100644 --- a/tokio/tests/tcp_peek.rs +++ b/tokio/tests/tcp_peek.rs @@ -17,7 +17,7 @@ async fn peek() { let left = net::TcpStream::connect(&addr).unwrap(); let mut right = t.join().unwrap(); - right.write(&[1, 2, 3, 4]).unwrap(); + let _ = right.write(&[1, 2, 3, 4]).unwrap(); let mut left: TcpStream = left.try_into().unwrap(); let mut buf = [0u8; 16];