diff --git a/quicr/Cargo.toml b/quicr/Cargo.toml index b86f22067..10519e31d 100644 --- a/quicr/Cargo.toml +++ b/quicr/Cargo.toml @@ -28,7 +28,7 @@ bytes = "0.4.7" rand = "0.4" [dev-dependencies] -tokio = "0.1.5" +tokio = "0.1.6" slog-term = "2" url = "1.7" structopt = "0.2.7" diff --git a/quicr/examples/client.rs b/quicr/examples/client.rs index e77262673..e9f8e68e7 100644 --- a/quicr/examples/client.rs +++ b/quicr/examples/client.rs @@ -1,5 +1,4 @@ extern crate tokio; -extern crate tokio_timer; extern crate quicr; #[macro_use] extern crate failure; @@ -17,7 +16,7 @@ use std::time::{Instant, Duration}; use std::path::PathBuf; use futures::Future; -use tokio::executor::current_thread::CurrentThread; +use tokio::runtime::current_thread::Runtime; use url::Url; use structopt::StructOpt; @@ -55,14 +54,10 @@ fn run(log: Logger, options: Opt) -> Result<()> { let url = options.url; let remote = url.with_default_port(|_| Ok(4433))?.to_socket_addrs()?.next().ok_or(format_err!("couldn't resolve to an address"))?; - let reactor = tokio::reactor::Reactor::new()?; - let handle = reactor.handle(); - let timer = tokio_timer::Timer::new(reactor); + let mut runtime = Runtime::new()?; let mut builder = quicr::Endpoint::new(); - builder.reactor(&handle) - .timer(timer.handle()) - .logger(log.clone()) + builder.logger(log.clone()) .config(quicr::Config { protocols: vec![b"hq-11"[..].into()], keylog: options.keylog, @@ -70,13 +65,11 @@ fn run(log: Logger, options: Opt) -> Result<()> { ..quicr::Config::default() }); let (endpoint, driver, _) = builder.bind("[::]:0")?; + runtime.spawn(driver.map_err(|e| eprintln!("IO error: {}", e))); - let mut executor = CurrentThread::new_with_park(timer); let request = format!("GET {}\r\n", url.path()); - let start = Instant::now(); - executor.spawn(driver.map_err(|e| eprintln!("IO error: {}", e))); - executor.block_on( + runtime.block_on( endpoint.connect(&remote, url.host_str().map(|x| x.as_bytes())) .map_err(|e| format_err!("failed to connect: {}", e)) .and_then(move |(conn, _)| { @@ -84,6 +77,7 @@ fn run(log: Logger, options: Opt) -> Result<()> { let stream = conn.open_bi(); stream.map_err(|e| format_err!("failed to open stream: {}", e)) .and_then(move |stream| { + eprintln!("stream opened at {}", duration_secs(&start.elapsed())); tokio::io::write_all(stream, request.as_bytes().to_owned()).map_err(|e| format_err!("failed to send request: {}", e)) }) .and_then(|(stream, _)| tokio::io::shutdown(stream).map_err(|e| format_err!("failed to shutdown stream: {}", e))) @@ -102,7 +96,7 @@ fn run(log: Logger, options: Opt) -> Result<()> { }) .map(|()| eprintln!("drained")) }) - ).map_err(|e| e.into_inner().unwrap())?; + )?; Ok(()) } diff --git a/quicr/examples/server.rs b/quicr/examples/server.rs index 7e84b7427..ea78bfe8e 100644 --- a/quicr/examples/server.rs +++ b/quicr/examples/server.rs @@ -1,5 +1,4 @@ extern crate tokio; -extern crate tokio_timer; extern crate quicr; #[macro_use] extern crate failure; @@ -19,7 +18,8 @@ use std::rc::Rc; use std::ascii; use futures::{Future, Stream}; -use tokio::executor::current_thread::{self, CurrentThread}; +use tokio::executor::current_thread; +use tokio::runtime::current_thread::Runtime; use failure::{ResultExt, Fail}; use structopt::StructOpt; @@ -84,14 +84,10 @@ fn run(log: Logger, options: Opt) -> Result<()> { let root = Rc::new(options.root); if !root.exists() { bail!("root path does not exist"); } - let reactor = tokio::reactor::Reactor::new()?; - let handle = reactor.handle(); - let timer = tokio_timer::Timer::new(reactor); + let mut runtime = Runtime::new()?; let mut builder = quicr::Endpoint::new(); - builder.reactor(&handle) - .timer(timer.handle()) - .logger(log.clone()) + builder.logger(log.clone()) .config(quicr::Config { protocols: vec![b"hq-11"[..].into()], max_remote_bi_streams: 64, @@ -116,9 +112,8 @@ fn run(log: Logger, options: Opt) -> Result<()> { } let (_, driver, incoming) = builder.bind("[::]:4433")?; - let mut executor = CurrentThread::new_with_park(timer); - executor.spawn(incoming.for_each(move |conn| { + runtime.spawn(incoming.for_each(move |conn| { let quicr::NewConnection { incoming, connection } = conn; let log = log.new(o!("local_id" => format!("{}", connection.local_id()))); info!(log, "got connection"; @@ -168,7 +163,7 @@ fn run(log: Logger, options: Opt) -> Result<()> { Ok(()) })); - executor.block_on(driver).map_err(|e| e.into_inner().unwrap())?; + runtime.block_on(driver)?; Ok(()) } diff --git a/quicr/src/lib.rs b/quicr/src/lib.rs index e30d88f56..49e779535 100644 --- a/quicr/src/lib.rs +++ b/quicr/src/lib.rs @@ -7,30 +7,21 @@ //! The entry point of this crate is the [`Endpoint`](struct.Endpoint.html). //! //! The futures and streams defined in this crate are not `Send` because they necessarily share state with eachother. As -//! a result, they must be spawned using an executor that operates on the same thread that they were constructed, such -//! as with `tokio::executor::current_thread`. The standard tokio runtime offloads futures to a threadpool, so its -//! components must instead be pieced together by hand as follows: +//! a result, they must be spawned on a single-threaded tokio runtime. //! //! ``` //! extern crate tokio; -//! extern crate tokio_timer; //! extern crate quicr; //! extern crate futures; //! //! use futures::Future; //! //! fn main() { -//! let reactor = tokio::reactor::Reactor::new().unwrap(); -//! let reactor_handle = reactor.handle(); -//! let timer = tokio_timer::Timer::new(reactor); -//! let timer_handle = timer.handle(); -//! let mut executor = tokio::executor::current_thread::CurrentThread::new_with_park(timer); -//! +//! let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap(); //! let mut builder = quicr::Endpoint::new(); -//! builder.reactor(&reactor_handle).timer(timer_handle); +//! // //! let (endpoint, driver, _) = builder.bind("[::]:0").unwrap(); -//! -//! executor.spawn(driver.map_err(|e| panic!("IO error: {}", e))); +//! runtime.spawn(driver.map_err(|e| panic!("IO error: {}", e))); //! // ... //! } //! ``` @@ -62,7 +53,7 @@ use std::borrow::Cow; use tokio_udp::UdpSocket; use tokio_io::{AsyncRead, AsyncWrite}; -use tokio_timer::{Delay, timer}; +use tokio_timer::Delay; use slog::Logger; use futures::{Future, Poll, Async}; use futures::Stream as FuturesStream; @@ -111,7 +102,6 @@ impl From for Error { struct EndpointInner { log: Logger, - timer: timer::Handle, socket: UdpSocket, inner: quicr::Endpoint, outgoing: VecDeque<(SocketAddrV6, Box<[u8]>)>, @@ -202,7 +192,6 @@ pub type Incoming = mpsc::UnboundedReceiver; /// A helper for constructing an `Endpoint`. pub struct EndpointBuilder<'a> { reactor: Option<&'a tokio_reactor::Handle>, - timer: Option, logger: Logger, listen: Option, config: Config, @@ -213,7 +202,6 @@ pub struct EndpointBuilder<'a> { #[allow(missing_docs)] impl<'a> EndpointBuilder<'a> { pub fn reactor(&mut self, handle: &'a tokio_reactor::Handle) -> &mut Self { self.reactor = Some(handle); self } - pub fn timer(&mut self, handle: timer::Handle) -> &mut Self { self.timer = Some(handle); self } pub fn logger(&mut self, logger: Logger) -> &mut Self { self.logger = logger; self } pub fn config(&mut self, config: Config) -> &mut Self { self.config = config; self } @@ -265,7 +253,6 @@ impl<'a> EndpointBuilder<'a> { let socket = UdpSocket::from_std(socket, &reactor).map_err(Error::Socket)?; let (send, recv) = mpsc::unbounded(); let rc = Rc::new(RefCell::new(EndpointInner { - timer: self.timer.unwrap_or_else(|| timer::Handle::current()), log: self.logger.clone(), socket: socket, inner: quicr::Endpoint::new(self.logger, self.config, cert_config, self.listen)?, @@ -293,7 +280,6 @@ impl Endpoint { /// Begin constructing an `Endpoint` pub fn new<'a>() -> EndpointBuilder<'a> { EndpointBuilder { reactor: None, - timer: None, logger: Logger::root(slog::Discard, o!()), listen: None, config: Config::default(), @@ -451,7 +437,7 @@ impl Future for Driver { endpoint.timers.push(Timer { conn: connection, ty: timer, - delay: endpoint.timer.delay(instant), + delay: Delay::new(instant), cancel: None, }); } @@ -474,7 +460,7 @@ impl Future for Driver { endpoint.timers.push(Timer { conn: connection, ty: timer, - delay: endpoint.timer.delay(instant), + delay: Delay::new(instant), cancel: Some(recv), }); }