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

chore: remove benches and fix/work around clippy lints (#1952)

Browse Source
This commit is contained in:
Artem Vorotnikov
2019-12-13 22:01:47 -08:00
committed by Carl Lerche
parent 91ecb4b4c2
commit d593c5b051
34 changed files with 89 additions and 1041 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/chat.rs
+3 -1
View File
@@ -49,7 +49,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
// 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.
//
examples/connect.rs
+4 -5
View File
@@ -36,10 +36,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
};
// 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::<SocketAddr>()?;
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<Option<Vec<u8>>> {
if buf.len() > 0 {
if !buf.is_empty() {
let len = buf.len();
Ok(Some(buf.split_to(len).into_iter().collect()))
} else {
examples/echo-udp.rs
+3 -1
View File
@@ -51,7 +51,9 @@ impl Server {
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
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()?);
examples/echo.rs
+3 -1
View File
@@ -33,7 +33,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
// 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
examples/print_each_packet.rs
+3 -1
View File
@@ -65,7 +65,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
// 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
examples/proxy.rs
+6 -2
View File
@@ -32,8 +32,12 @@ use std::error::Error;
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
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);
examples/tinydb.rs
+9 -10
View File
@@ -84,7 +84,9 @@ enum Response {
async fn main() -> Result<(), Box<dyn Error>> {
// 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<Database>) -> Response {
impl Request {
fn parse(input: &str) -> Result<Request, String> {
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()),
}
}
}
examples/tinyhttp.rs
+3 -1
View File
@@ -27,7 +27,9 @@ use tokio_util::codec::{Decoder, Encoder, Framed};
async fn main() -> Result<(), Box<dyn Error>> {
// 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);
examples/udp-client.rs
+1 -1
View File
@@ -44,7 +44,7 @@ fn get_stdin_data() -> Result<Vec<u8>, Box<dyn std::error::Error>> {
async fn main() -> Result<(), Box<dyn Error>> {
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.
examples/udp-codec.rs
+3 -1
View File
@@ -22,7 +22,9 @@ use std::time::Duration;
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
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?;
tokio-test/src/task.rs
+2
View File
@@ -1,5 +1,7 @@
//! Futures task based helpers
#![allow(clippy::mutex_atomic)]
use futures_core::Stream;
use std::future::Future;
use std::mem;
tokio-test/tests/block_on.rs
+4 -6
View File
@@ -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;
});
}
tokio-util/tests/framed_write.rs
+1 -1
View File
@@ -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[..]);
tokio/benches/latency.rs
-114
View File
@@ -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<u8>,
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::<SocketAddr>().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();
}
tokio/benches/mio-ops.rs
-57
View File
@@ -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();
});
}
tokio/benches/mpsc.rs
-270
View File
@@ -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::<Medium>(1_000));
})
}
#[bench]
fn unbounded_new_medium(b: &mut Bencher) {
b.iter(|| {
let _ = test::black_box(&unbounded_channel::<Medium>());
})
}
#[bench]
fn bounded_new_large(b: &mut Bencher) {
b.iter(|| {
let _ = test::black_box(&channel::<Large>(1_000));
})
}
#[bench]
fn unbounded_new_large(b: &mut Bencher) {
b.iter(|| {
let _ = test::black_box(&unbounded_channel::<Large>());
})
}
#[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::<Large>(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::<i32>(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::<i32>(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::<i32>(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::<i32>();
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::<i32>();
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::<Large>(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::<Sender<i32>>();
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::<i32>(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::<Sender<i32>>();
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::<i32>(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();
}
}
tokio/benches/oneshot.rs
-120
View File
@@ -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::<i32>());
})
}
#[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<F: Future>(mut f: F) -> Result<F::Item, F::Error> {
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();
}
tokio/benches/tcp.rs
-257
View File
@@ -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);
}
}
}
tokio/benches/thread_pool.rs
-153
View File
@@ -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();
});
}
tokio/src/lib.rs
+1
View File
@@ -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,
tokio/src/macros/assert.rs
+2 -3
View File
@@ -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);
}
}};
}
tokio/src/runtime/builder.rs
+1 -1
View File
@@ -430,7 +430,7 @@ cfg_rt_threaded! {
io_handle,
time_handle,
clock,
blocking_spawner: blocking_spawner.clone(),
blocking_spawner,
},
blocking_pool,
})
tokio/src/runtime/shell.rs
+2
View File
@@ -1,3 +1,5 @@
#![allow(clippy::redundant_clone)]
use crate::park::Park;
use crate::runtime::enter;
use crate::runtime::time;
tokio/src/runtime/thread_pool/tests/queue.rs
+5 -9
View File
@@ -41,9 +41,8 @@ macro_rules! assert_steal {
macro_rules! assert_empty {
($q:expr) => {{
let q: &mut queue::Worker<Noop> = &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<Noop>) -> 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;
}
}
tokio/src/signal/registry.rs
+2
View File
@@ -1,3 +1,5 @@
#![allow(clippy::unit_arg)]
use crate::signal::os::{OsExtraData, OsStorage};
use crate::sync::mpsc::Sender;
tokio/src/time/tests/test_queue.rs
+21 -23
View File
@@ -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");
})
}
tokio/tests/io_async_read.rs
+1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::transmute_ptr_to_ptr)]
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
tokio/tests/process_issue_42.rs
+2 -2
View File
@@ -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())
tokio/tests/rt_common.rs
+1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::needless_range_loop)]
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
tokio/tests/support/mock_file.rs
+2
View File
@@ -1,3 +1,5 @@
#![allow(clippy::unnecessary_operation)]
use std::collections::VecDeque;
use std::fmt;
use std::fs::{Metadata, Permissions};
tokio/tests/sync_barrier.rs
+1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::unnecessary_operation)]
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
tokio/tests/sync_mpsc.rs
+1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::redundant_clone)]
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
tokio/tests/sync_watch.rs
+1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::cognitive_complexity)]
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
tokio/tests/tcp_peek.rs
+1 -1
View File
@@ -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];