From 89fcc96dd44bff0ba85432d96a3a8f5b20adc94e Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Sun, 5 Feb 2017 17:06:57 -0800 Subject: [PATCH] Migrate to using tokio-io Deprecate the existing `io` module in this crate entirely. More details coming soon! Closes #61 --- Cargo.toml | 7 ++- examples/chat.rs | 11 ++-- examples/connect.rs | 35 ++++++++---- examples/echo.rs | 6 +- examples/hello.rs | 5 +- examples/sink.rs | 5 +- src/channel.rs | 5 ++ src/io/frame.rs | 5 ++ src/io/mod.rs | 12 ++-- src/io/split.rs | 7 +-- src/lib.rs | 10 ++++ src/net/tcp.rs | 106 +++++++++++++++++++++++++++++++----- src/net/udp/frame.rs | 5 ++ src/net/udp/mod.rs | 8 +-- src/reactor/io_token.rs | 4 +- src/reactor/mod.rs | 81 +++++++++++++++++++-------- src/reactor/poll_evented.rs | 50 ++++++++++++----- tests/buffered.rs | 7 ++- tests/chain.rs | 3 +- tests/echo.rs | 6 +- tests/limit.rs | 3 +- tests/line-frames.rs | 40 +++++++++----- tests/pipe-hup.rs | 15 +++-- tests/spawn.rs | 30 +++++----- tests/stream-buffered.rs | 6 +- 25 files changed, 340 insertions(+), 132 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index cef1c19ce..870eedd97 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,11 +17,14 @@ travis-ci = { repository = "tokio-rs/tokio-core" } appveyor = { repository = "alexcrichton/tokio-core" } [dependencies] -futures = "0.1.9" +bytes = "0.4" log = "0.3" -mio = "0.6.4" +mio = "0.6.5" scoped-tls = "0.1.0" slab = "0.3" +iovec = "0.1" +tokio-io = "0.1" +futures = "0.1.11" [dev-dependencies] env_logger = { version = "0.3", default-features = false } diff --git a/examples/chat.rs b/examples/chat.rs index e7b472dd8..267e0aa63 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -17,8 +17,9 @@ //! connected clients they'll all join the same room and see everyone else's //! messages. -extern crate tokio_core; extern crate futures; +extern crate tokio_core; +extern crate tokio_io; use std::collections::HashMap; use std::rc::Rc; @@ -27,12 +28,12 @@ use std::iter; use std::env; use std::io::{Error, ErrorKind, BufReader}; +use futures::Future; +use futures::stream::{self, Stream}; use tokio_core::net::TcpListener; use tokio_core::reactor::Core; -use tokio_core::io::{self, Io}; - -use futures::stream::{self, Stream}; -use futures::Future; +use tokio_io::io; +use tokio_io::AsyncRead; fn main() { let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); diff --git a/examples/connect.rs b/examples/connect.rs index 93b6d5d4e..a167e006e 100644 --- a/examples/connect.rs +++ b/examples/connect.rs @@ -10,17 +10,21 @@ extern crate futures; extern crate tokio_core; +extern crate tokio_io; +extern crate bytes; use std::env; use std::io::{self, Read, Write}; use std::net::SocketAddr; use std::thread; -use futures::{Sink, Future, Stream}; +use bytes::{BufMut, BytesMut}; use futures::sync::mpsc; -use tokio_core::reactor::Core; -use tokio_core::io::{Io, EasyBuf, Codec}; +use futures::{Sink, Future, Stream}; use tokio_core::net::TcpStream; +use tokio_core::reactor::Core; +use tokio_io::AsyncRead; +use tokio_io::codec::{Encoder, Decoder}; fn main() { // Parse what address we're going to connect to @@ -63,7 +67,7 @@ fn main() { let (sink, stream) = stream.framed(Bytes).split(); let send_stdin = stdin_rx.forward(sink); let write_stdout = stream.for_each(move |buf| { - stdout.write_all(buf.as_slice()) + stdout.write_all(&buf) }); send_stdin.map(|_| ()) @@ -83,21 +87,30 @@ fn main() { /// data into the output location without looking at it. struct Bytes; -impl Codec for Bytes { - type In = EasyBuf; - type Out = Vec; +impl Decoder for Bytes { + type Item = BytesMut; + type Error = io::Error; - fn decode(&mut self, buf: &mut EasyBuf) -> io::Result> { + fn decode(&mut self, buf: &mut BytesMut) -> io::Result> { if buf.len() > 0 { let len = buf.len(); - Ok(Some(buf.drain_to(len))) + Ok(Some(buf.split_to(len))) } else { Ok(None) } } - fn encode(&mut self, data: Vec, buf: &mut Vec) -> io::Result<()> { - buf.extend(data); + fn decode_eof(&mut self, buf: &mut BytesMut) -> io::Result> { + self.decode(buf) + } +} + +impl Encoder for Bytes { + type Item = Vec; + type Error = io::Error; + + fn encode(&mut self, data: Vec, buf: &mut BytesMut) -> io::Result<()> { + buf.put(&data[..]); Ok(()) } } diff --git a/examples/echo.rs b/examples/echo.rs index 45ced33e6..80e73ea74 100644 --- a/examples/echo.rs +++ b/examples/echo.rs @@ -19,13 +19,15 @@ extern crate futures; extern crate tokio_core; +extern crate tokio_io; use std::env; use std::net::SocketAddr; use futures::Future; use futures::stream::Stream; -use tokio_core::io::{copy, Io}; +use tokio_io::AsyncRead; +use tokio_io::io::copy; use tokio_core::net::TcpListener; use tokio_core::reactor::Core; @@ -93,7 +95,7 @@ fn main() { // information. let msg = amt.then(move |result| { match result { - Ok(amt) => println!("wrote {} bytes to {}", amt, addr), + Ok((amt, _, _)) => println!("wrote {} bytes to {}", amt, addr), Err(e) => println!("error on {}: {}", addr, e), } diff --git a/examples/hello.rs b/examples/hello.rs index df83115bf..a22517a5a 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -11,9 +11,10 @@ //! //! You should see `Hello!` printed out and then the `nc` program will exit. +extern crate env_logger; extern crate futures; extern crate tokio_core; -extern crate env_logger; +extern crate tokio_io; use std::env; use std::net::SocketAddr; @@ -35,7 +36,7 @@ fn main() { let clients = listener.incoming(); let welcomes = clients.and_then(|(socket, _peer_addr)| { - tokio_core::io::write_all(socket, b"Hello!\n") + tokio_io::io::write_all(socket, b"Hello!\n") }); let server = welcomes.for_each(|(_socket, _welcome)| { Ok(()) diff --git a/examples/sink.rs b/examples/sink.rs index 2893adc64..2998b269f 100644 --- a/examples/sink.rs +++ b/examples/sink.rs @@ -18,6 +18,7 @@ extern crate env_logger; extern crate futures; extern crate tokio_core; +extern crate tokio_io; use std::env; use std::iter; @@ -25,7 +26,7 @@ use std::net::SocketAddr; use futures::Future; use futures::stream::{self, Stream}; -use tokio_core::io::IoFuture; +use tokio_io::IoFuture; use tokio_core::net::{TcpListener, TcpStream}; use tokio_core::reactor::Core; @@ -51,6 +52,6 @@ fn write(socket: TcpStream) -> IoFuture<()> { static BUF: &'static [u8] = &[0; 64 * 1024]; let iter = iter::repeat(()).map(|()| Ok(())); stream::iter(iter).fold(socket, |socket, ()| { - tokio_core::io::write_all(socket, BUF).map(|(socket, _)| socket) + tokio_io::io::write_all(socket, BUF).map(|(socket, _)| socket) }).map(|_| ()).boxed() } diff --git a/src/channel.rs b/src/channel.rs index f70080f1c..6251afb49 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -5,6 +5,7 @@ #![deprecated(since = "0.1.1", note = "use `futures::sync::mpsc` instead")] #![allow(deprecated)] +#![cfg(feature = "with-deprecated")] use std::io; use std::sync::mpsc::TryRecvError; @@ -95,6 +96,10 @@ impl Sink for Sender { fn poll_complete(&mut self) -> Poll<(), io::Error> { Ok(().into()) } + + fn close(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } } impl Clone for Sender { diff --git a/src/io/frame.rs b/src/io/frame.rs index ea82f67af..821dc3cfb 100644 --- a/src/io/frame.rs +++ b/src/io/frame.rs @@ -404,6 +404,11 @@ impl Sink for Framed { trace!("framed transport flushed"); return Ok(Async::Ready(())); } + + fn close(&mut self) -> Poll<(), io::Error> { + try_ready!(self.poll_complete()); + Ok(().into()) + } } pub fn framed(io: T, codec: C) -> Framed { diff --git a/src/io/mod.rs b/src/io/mod.rs index ba0c14e6f..c3f7e7d11 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -9,12 +9,14 @@ //! [found online]: https://tokio.rs/docs/getting-started/core/ //! [low level details]: https://tokio.rs/docs/going-deeper/core-low-level/ +#![deprecated(note = "moved to the `tokio-io` crate")] + use std::io; -use futures::{BoxFuture, Async, Poll}; +use futures::{Async, Poll}; +use futures::future::BoxFuture; use futures::stream::BoxStream; - -use mio::IoVec; +use iovec::IoVec; /// A convenience typedef around a `Future` whose error component is `io::Error` pub type IoFuture = BoxFuture; @@ -138,7 +140,7 @@ pub trait Io: io::Read + io::Write { if bufs.is_empty() { Ok(0) } else { - self.read(bufs[0].as_mut_bytes()) + self.read(&mut bufs[0]) } } @@ -163,7 +165,7 @@ pub trait Io: io::Read + io::Write { if bufs.is_empty() { Ok(0) } else { - self.write(bufs[0].as_bytes()) + self.write(&bufs[0]) } } diff --git a/src/io/split.rs b/src/io/split.rs index a012ed69a..22f3d63b4 100644 --- a/src/io/split.rs +++ b/src/io/split.rs @@ -2,7 +2,6 @@ use std::io::{self, Read, Write}; use futures::Async; use futures::sync::BiLock; -use mio; use io::Io; @@ -47,7 +46,7 @@ impl Read for ReadHalf { fn read(&mut self, buf: &mut [u8]) -> io::Result { match self.handle.poll_lock() { Async::Ready(mut l) => l.read(buf), - Async::NotReady => Err(mio::would_block()), + Async::NotReady => Err(::would_block()), } } } @@ -56,14 +55,14 @@ impl Write for WriteHalf { fn write(&mut self, buf: &[u8]) -> io::Result { match self.handle.poll_lock() { Async::Ready(mut l) => l.write(buf), - Async::NotReady => Err(mio::would_block()), + Async::NotReady => Err(::would_block()), } } fn flush(&mut self) -> io::Result<()> { match self.handle.poll_lock() { Async::Ready(mut l) => l.flush(), - Async::NotReady => Err(mio::would_block()), + Async::NotReady => Err(::would_block()), } } } diff --git a/src/lib.rs b/src/lib.rs index cccdef749..b34ff9825 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -90,9 +90,13 @@ #![doc(html_root_url = "https://docs.rs/tokio-core/0.1")] #![deny(missing_docs)] +extern crate bytes; +#[macro_use] extern crate futures; +extern crate iovec; extern crate mio; extern crate slab; +extern crate tokio_io; #[macro_use] extern crate scoped_tls; @@ -108,3 +112,9 @@ mod heap; pub mod channel; pub mod net; pub mod reactor; + +use std::io as sio; + +fn would_block() -> sio::Error { + sio::Error::new(sio::ErrorKind::WouldBlock, "would block") +} diff --git a/src/net/tcp.rs b/src/net/tcp.rs index 192db9d6f..caf466808 100644 --- a/src/net/tcp.rs +++ b/src/net/tcp.rs @@ -3,12 +3,14 @@ use std::io::{self, Read, Write}; use std::mem; use std::net::{self, SocketAddr, Shutdown}; +use bytes::{Buf, BufMut}; use futures::stream::Stream; use futures::sync::oneshot; use futures::{Future, Poll, Async}; +use iovec::IoVec; use mio; +use tokio_io::{AsyncRead, AsyncWrite}; -use io::{Io, IoFuture}; use reactor::{Handle, PollEvented}; /// An I/O object representing a TCP socket listening for incoming connections. @@ -60,7 +62,7 @@ impl TcpListener { match pending.poll().expect("shouldn't be canceled") { Async::NotReady => { self.pending_accept = Some(pending); - return Err(mio::would_block()) + return Err(::would_block()) }, Async::Ready(r) => return r, } @@ -94,7 +96,7 @@ impl TcpListener { .map(move |io| { (TcpStream { io: io }, addr) }); - tx.complete(res); + drop(tx.send(res)); Ok(()) }); self.pending_accept = Some(rx); @@ -299,7 +301,8 @@ impl TcpStream { /// (perhaps to `INADDR_ANY`) before this method is called. pub fn connect_stream(stream: net::TcpStream, addr: &SocketAddr, - handle: &Handle) -> IoFuture { + handle: &Handle) + -> Box + Send> { let state = match mio::tcp::TcpStream::connect_stream(stream, addr) { Ok(tcp) => TcpStream::new(tcp, handle), Err(e) => TcpStreamNewState::Error(e), @@ -425,7 +428,28 @@ impl Write for TcpStream { } } -impl Io for TcpStream { +impl AsyncRead for TcpStream { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } + + fn read_buf(&mut self, buf: &mut B) -> Poll { + <&TcpStream>::read_buf(&mut &*self, buf) + } +} + +impl AsyncWrite for TcpStream { + fn shutdown(&mut self) -> Poll<(), io::Error> { + <&TcpStream>::shutdown(&mut &*self) + } + + fn write_buf(&mut self, buf: &mut B) -> Poll { + <&TcpStream>::write_buf(&mut &*self, buf) + } +} + +#[allow(deprecated)] +impl ::io::Io for TcpStream { fn poll_read(&mut self) -> Async<()> { ::poll_read(self) } @@ -434,29 +458,26 @@ impl Io for TcpStream { ::poll_write(self) } - fn read_vec(&mut self, bufs: &mut [&mut mio::IoVec]) -> io::Result { - if let Async::NotReady = self.poll_read() { - return Err(mio::would_block()) + fn read_vec(&mut self, bufs: &mut [&mut IoVec]) -> io::Result { + if let Async::NotReady = ::poll_read(self) { + return Err(::would_block()) } let r = self.io.get_ref().read_bufs(bufs); if is_wouldblock(&r) { self.io.need_read(); } return r - - } - fn write_vec(&mut self, bufs: &[&mio::IoVec]) -> io::Result { - if let Async::NotReady = self.poll_write() { - return Err(mio::would_block()) + fn write_vec(&mut self, bufs: &[&IoVec]) -> io::Result { + if let Async::NotReady = ::poll_write(self) { + return Err(::would_block()) } let r = self.io.get_ref().write_bufs(bufs); if is_wouldblock(&r) { self.io.need_write(); } return r - } } @@ -483,7 +504,62 @@ impl<'a> Write for &'a TcpStream { } } -impl<'a> Io for &'a TcpStream { +impl<'a> AsyncRead for &'a TcpStream { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } + + fn read_buf(&mut self, buf: &mut B) -> Poll { + if let Async::NotReady = ::poll_read(self) { + return Err(::would_block()) + } + let mut bufs: [_; 16] = Default::default(); + unsafe { + let n = buf.bytes_vec_mut(&mut bufs); + match self.io.get_ref().read_bufs(&mut bufs[..n]) { + Ok(n) => { + buf.advance_mut(n); + Ok(Async::Ready(n)) + } + Err(e) => { + if e.kind() == io::ErrorKind::WouldBlock { + self.io.need_write(); + } + Err(e) + } + } + } + } +} + +impl<'a> AsyncWrite for &'a TcpStream { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } + + fn write_buf(&mut self, buf: &mut B) -> Poll { + if let Async::NotReady = ::poll_write(self) { + return Err(::would_block()) + } + let mut bufs: [_; 16] = Default::default(); + let n = buf.bytes_vec(&mut bufs); + match self.io.get_ref().write_bufs(&bufs[..n]) { + Ok(n) => { + buf.advance(n); + Ok(Async::Ready(n)) + } + Err(e) => { + if e.kind() == io::ErrorKind::WouldBlock { + self.io.need_write(); + } + Err(e) + } + } + } +} + +#[allow(deprecated)] +impl<'a> ::io::Io for &'a TcpStream { fn poll_read(&mut self) -> Async<()> { ::poll_read(self) } diff --git a/src/net/udp/frame.rs b/src/net/udp/frame.rs index 566d9058a..28705f959 100644 --- a/src/net/udp/frame.rs +++ b/src/net/udp/frame.rs @@ -111,6 +111,11 @@ impl Sink for UdpFramed { "failed to write entire datagram to socket")) } } + + fn close(&mut self) -> Poll<(), io::Error> { + try_ready!(self.poll_complete()); + Ok(().into()) + } } pub fn new(socket: UdpSocket, codec: C) -> UdpFramed { diff --git a/src/net/udp/mod.rs b/src/net/udp/mod.rs index 980d6f29a..7592ef21e 100644 --- a/src/net/udp/mod.rs +++ b/src/net/udp/mod.rs @@ -101,13 +101,13 @@ impl UdpSocket { /// documentation for concrete examples. pub fn send_to(&self, buf: &[u8], target: &SocketAddr) -> io::Result { if let Async::NotReady = self.io.poll_write() { - return Err(mio::would_block()) + return Err(::would_block()) } match self.io.get_ref().send_to(buf, target) { Ok(Some(n)) => Ok(n), Ok(None) => { self.io.need_write(); - Err(mio::would_block()) + Err(::would_block()) } Err(e) => Err(e), } @@ -144,13 +144,13 @@ impl UdpSocket { /// read and the address from whence the data came. pub fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { if let Async::NotReady = self.io.poll_read() { - return Err(mio::would_block()) + return Err(::would_block()) } match self.io.get_ref().recv_from(buf) { Ok(Some(n)) => Ok(n), Ok(None) => { self.io.need_read(); - Err(mio::would_block()) + Err(::would_block()) } Err(e) => Err(e), } diff --git a/src/reactor/io_token.rs b/src/reactor/io_token.rs index 94782e5f5..e8c4880b1 100644 --- a/src/reactor/io_token.rs +++ b/src/reactor/io_token.rs @@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::io; use futures::task; -use mio; +use mio::event::Evented; use reactor::{Message, Remote, Handle, Direction}; @@ -31,7 +31,7 @@ impl IoToken { /// The returned future will panic if the event loop this handle is /// associated with has gone away, or if there is an error communicating /// with the event loop. - pub fn new(source: &mio::Evented, handle: &Handle) -> io::Result { + pub fn new(source: &Evented, handle: &Handle) -> io::Result { match handle.inner.upgrade() { Some(inner) => { let (ready, token) = try!(inner.borrow_mut().add_source(source)); diff --git a/src/reactor/mod.rs b/src/reactor/mod.rs index c2d80de7a..b24e5f38a 100644 --- a/src/reactor/mod.rs +++ b/src/reactor/mod.rs @@ -14,11 +14,13 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering}; use std::time::{Instant, Duration}; -use futures::{self, Future, IntoFuture, Async}; +use futures::{Future, IntoFuture, Async}; +use futures::future; use futures::executor::{self, Spawn, Unpark}; use futures::sync::mpsc; use futures::task::Task; use mio; +use mio::event::Evented; use slab::Slab; use heap::{Heap, Slot}; @@ -153,15 +155,17 @@ impl Core { /// creation. pub fn new() -> io::Result { let io = try!(mio::Poll::new()); - let future_pair = mio::Registration::new(&io, - TOKEN_FUTURE, - mio::Ready::readable(), - mio::PollOpt::level()); + let future_pair = mio::Registration::new2(); + try!(io.register(&future_pair.0, + TOKEN_FUTURE, + mio::Ready::readable(), + mio::PollOpt::level())); let (tx, rx) = mpsc::unbounded(); - let channel_pair = mio::Registration::new(&io, - TOKEN_MESSAGES, - mio::Ready::readable(), - mio::PollOpt::level()); + let channel_pair = mio::Registration::new2(); + try!(io.register(&channel_pair.0, + TOKEN_MESSAGES, + mio::Ready::readable(), + mio::PollOpt::level())); let rx_readiness = Arc::new(MySetReadiness(channel_pair.1)); rx_readiness.unpark(); @@ -296,16 +300,16 @@ impl Core { for i in 0..self.events.len() { let event = self.events.get(i).unwrap(); let token = event.token(); - trace!("event {:?} {:?}", event.kind(), event.token()); + trace!("event {:?} {:?}", event.readiness(), event.token()); if token == TOKEN_MESSAGES { - self.rx_readiness.0.set_readiness(mio::Ready::none()).unwrap(); + self.rx_readiness.0.set_readiness(mio::Ready::empty()).unwrap(); CURRENT_LOOP.set(&self, || self.consume_queue()); } else if token == TOKEN_FUTURE { - self.future_readiness.0.set_readiness(mio::Ready::none()).unwrap(); + self.future_readiness.0.set_readiness(mio::Ready::empty()).unwrap(); fired = true; } else { - self.dispatch(token, event.kind()); + self.dispatch(token, event.readiness()); } } debug!("loop process - {} events, {:?}", amt, after_poll.elapsed()); @@ -326,7 +330,7 @@ impl Core { let mut writer = None; let mut inner = self.inner.borrow_mut(); if let Some(io) = inner.io_dispatch.get_mut(token) { - if ready.is_readable() || ready.is_hup() { + if ready.is_readable() || platform::is_hup(&ready) { reader = io.reader.take(); io.readiness.fetch_or(Readiness::Readable as usize, Ordering::Relaxed); @@ -353,7 +357,7 @@ impl Core { Some(slot) => (slot.spawn.take(), slot.wake.clone()), None => return, }; - wake.0.set_readiness(mio::Ready::none()).unwrap(); + wake.0.set_readiness(mio::Ready::empty()).unwrap(); let mut task = match task { Some(task) => task, None => return, @@ -455,7 +459,7 @@ impl fmt::Debug for Core { } impl Inner { - fn add_source(&mut self, source: &mio::Evented) + fn add_source(&mut self, source: &Evented) -> io::Result<(Arc, usize)> { debug!("adding a new I/O source"); let sched = ScheduledIo { @@ -470,12 +474,14 @@ impl Inner { let entry = self.io_dispatch.vacant_entry().unwrap(); try!(self.io.register(source, mio::Token(TOKEN_START + entry.index() * 2), - mio::Ready::readable() | mio::Ready::writable() | mio::Ready::hup(), + mio::Ready::readable() | + mio::Ready::writable() | + platform::hup(), mio::PollOpt::edge())); Ok((sched.readiness.clone(), entry.insert(sched).index())) } - fn deregister_source(&mut self, source: &mio::Evented) -> io::Result<()> { + fn deregister_source(&mut self, source: &Evented) -> io::Result<()> { self.io.deregister(source) } @@ -546,10 +552,12 @@ impl Inner { } let entry = self.task_dispatch.vacant_entry().unwrap(); let token = TOKEN_START + 2 * entry.index() + 1; - let pair = mio::Registration::new(&self.io, - mio::Token(token), - mio::Ready::readable(), - mio::PollOpt::level()); + let pair = mio::Registration::new2(); + self.io.register(&pair.0, + mio::Token(token), + mio::Ready::readable(), + mio::PollOpt::level()) + .expect("cannot fail future registration with mio"); let unpark = Arc::new(MySetReadiness(pair.1)); let entry = entry.insert(ScheduledTask { spawn: Some(executor::spawn(future)), @@ -688,7 +696,7 @@ impl Handle { where F: FnOnce() -> R + 'static, R: IntoFuture + 'static, { - self.spawn(futures::lazy(f)) + self.spawn(future::lazy(f)) } /// Return the ID of the represented Core @@ -742,3 +750,30 @@ impl FnBox for F { (*self)(lp) } } + +#[cfg(unix)] +mod platform { + use mio::Ready; + use mio::unix::UnixReady; + + pub fn is_hup(event: &Ready) -> bool { + UnixReady::from(*event).is_hup() + } + + pub fn hup() -> Ready { + UnixReady::hup().into() + } +} + +#[cfg(windows)] +mod platform { + use mio::Ready; + + pub fn is_hup(_event: &Ready) -> bool { + false + } + + pub fn hup() -> Ready { + Ready::empty() + } +} diff --git a/src/reactor/poll_evented.rs b/src/reactor/poll_evented.rs index a957d824a..5d30b5c09 100644 --- a/src/reactor/poll_evented.rs +++ b/src/reactor/poll_evented.rs @@ -10,10 +10,10 @@ use std::fmt; use std::io::{self, Read, Write}; use std::sync::atomic::{AtomicUsize, Ordering}; -use futures::Async; -use mio; +use futures::{Async, Poll}; +use mio::event::Evented; +use tokio_io::{AsyncRead, AsyncWrite}; -use io::Io; use reactor::{Handle, Remote}; use reactor::Readiness::*; use reactor::io_token::IoToken; @@ -45,7 +45,7 @@ pub struct PollEvented { io: E, } -impl fmt::Debug for PollEvented { +impl fmt::Debug for PollEvented { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { f.debug_struct("PollEvented") .field("io", &self.io) @@ -53,7 +53,7 @@ impl fmt::Debug for PollEvented { } } -impl PollEvented { +impl PollEvented { /// Creates a new readiness stream associated with the provided /// `loop_handle` and for the given `source`. /// @@ -193,7 +193,7 @@ impl PollEvented { impl Read for PollEvented { fn read(&mut self, buf: &mut [u8]) -> io::Result { if let Async::NotReady = self.poll_read() { - return Err(mio::would_block()) + return Err(::would_block()) } let r = self.get_mut().read(buf); if is_wouldblock(&r) { @@ -206,7 +206,7 @@ impl Read for PollEvented { impl Write for PollEvented { fn write(&mut self, buf: &[u8]) -> io::Result { if let Async::NotReady = self.poll_write() { - return Err(mio::would_block()) + return Err(::would_block()) } let r = self.get_mut().write(buf); if is_wouldblock(&r) { @@ -217,7 +217,7 @@ impl Write for PollEvented { fn flush(&mut self) -> io::Result<()> { if let Async::NotReady = self.poll_write() { - return Err(mio::would_block()) + return Err(::would_block()) } let r = self.get_mut().flush(); if is_wouldblock(&r) { @@ -227,7 +227,17 @@ impl Write for PollEvented { } } -impl Io for PollEvented { +impl AsyncRead for PollEvented { +} + +impl AsyncWrite for PollEvented { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } +} + +#[allow(deprecated)] +impl ::io::Io for PollEvented { fn poll_read(&mut self) -> Async<()> { >::poll_read(self) } @@ -242,7 +252,7 @@ impl<'a, E> Read for &'a PollEvented { fn read(&mut self, buf: &mut [u8]) -> io::Result { if let Async::NotReady = self.poll_read() { - return Err(mio::would_block()) + return Err(::would_block()) } let r = self.get_ref().read(buf); if is_wouldblock(&r) { @@ -257,7 +267,7 @@ impl<'a, E> Write for &'a PollEvented { fn write(&mut self, buf: &[u8]) -> io::Result { if let Async::NotReady = self.poll_write() { - return Err(mio::would_block()) + return Err(::would_block()) } let r = self.get_ref().write(buf); if is_wouldblock(&r) { @@ -268,7 +278,7 @@ impl<'a, E> Write for &'a PollEvented fn flush(&mut self) -> io::Result<()> { if let Async::NotReady = self.poll_write() { - return Err(mio::would_block()) + return Err(::would_block()) } let r = self.get_ref().flush(); if is_wouldblock(&r) { @@ -278,7 +288,21 @@ impl<'a, E> Write for &'a PollEvented } } -impl<'a, E> Io for &'a PollEvented +impl<'a, E> AsyncRead for &'a PollEvented + where &'a E: Read, +{ +} + +impl<'a, E> AsyncWrite for &'a PollEvented + where &'a E: Write, +{ + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } +} + +#[allow(deprecated)] +impl<'a, E> ::io::Io for &'a PollEvented where &'a E: Read + Write, { fn poll_read(&mut self) -> Async<()> { diff --git a/tests/buffered.rs b/tests/buffered.rs index 819fe4ed2..c24232408 100644 --- a/tests/buffered.rs +++ b/tests/buffered.rs @@ -1,6 +1,7 @@ +extern crate env_logger; extern crate futures; extern crate tokio_core; -extern crate env_logger; +extern crate tokio_io; use std::net::TcpStream; use std::thread; @@ -8,7 +9,7 @@ use std::io::{Read, Write, BufReader, BufWriter}; use futures::Future; use futures::stream::Stream; -use tokio_core::io::copy; +use tokio_io::io::copy; use tokio_core::net::TcpListener; use tokio_core::reactor::Core; @@ -55,7 +56,7 @@ fn echo_server() { copy(a, b) }); - let amt = t!(l.run(copied)); + let (amt, _, _) = t!(l.run(copied)); let (expected, t2) = t.join().unwrap(); let actual = t2.join().unwrap(); diff --git a/tests/chain.rs b/tests/chain.rs index ae91ca274..fd69e0b9f 100644 --- a/tests/chain.rs +++ b/tests/chain.rs @@ -1,5 +1,6 @@ extern crate futures; extern crate tokio_core; +extern crate tokio_io; use std::net::TcpStream; use std::thread; @@ -7,7 +8,7 @@ use std::io::{Write, Read}; use futures::Future; use futures::stream::Stream; -use tokio_core::io::read_to_end; +use tokio_io::io::read_to_end; use tokio_core::net::TcpListener; use tokio_core::reactor::Core; diff --git a/tests/echo.rs b/tests/echo.rs index 83c5a8390..7a1bf7ff1 100644 --- a/tests/echo.rs +++ b/tests/echo.rs @@ -1,6 +1,7 @@ extern crate env_logger; extern crate futures; extern crate tokio_core; +extern crate tokio_io; use std::io::{Read, Write}; use std::net::TcpStream; @@ -8,9 +9,10 @@ use std::thread; use futures::Future; use futures::stream::Stream; -use tokio_core::io::{copy, Io}; use tokio_core::net::TcpListener; use tokio_core::reactor::Core; +use tokio_io::AsyncRead; +use tokio_io::io::copy; macro_rules! t { ($e:expr) => (match $e { @@ -44,7 +46,7 @@ fn echo_server() { let halves = client.map(|s| s.0.split()); let copied = halves.and_then(|(a, b)| copy(a, b)); - let amt = t!(l.run(copied)); + let (amt, _, _) = t!(l.run(copied)); t.join().unwrap(); assert_eq!(amt, msg.len() as u64 * 1024); diff --git a/tests/limit.rs b/tests/limit.rs index dac363d1c..53bbbf7b7 100644 --- a/tests/limit.rs +++ b/tests/limit.rs @@ -1,5 +1,6 @@ extern crate futures; extern crate tokio_core; +extern crate tokio_io; use std::net::TcpStream; use std::thread; @@ -7,7 +8,7 @@ use std::io::{Write, Read}; use futures::Future; use futures::stream::Stream; -use tokio_core::io::read_to_end; +use tokio_io::io::read_to_end; use tokio_core::net::TcpListener; use tokio_core::reactor::Core; diff --git a/tests/line-frames.rs b/tests/line-frames.rs index d5c9a898b..3270d685b 100644 --- a/tests/line-frames.rs +++ b/tests/line-frames.rs @@ -1,35 +1,49 @@ -extern crate tokio_core; extern crate env_logger; extern crate futures; +extern crate tokio_core; +extern crate tokio_io; +extern crate bytes; use std::io; use std::net::Shutdown; +use bytes::{BytesMut, BufMut}; use futures::{Future, Stream, Sink}; -use tokio_core::io::{write_all, read, Codec, EasyBuf, Io}; use tokio_core::net::{TcpListener, TcpStream}; use tokio_core::reactor::Core; +use tokio_io::codec::{Encoder, Decoder}; +use tokio_io::io::{write_all, read}; +use tokio_io::AsyncRead; pub struct LineCodec; -impl Codec for LineCodec { - type In = EasyBuf; - type Out = EasyBuf; +impl Decoder for LineCodec { + type Item = BytesMut; + type Error = io::Error; - fn decode(&mut self, buf: &mut EasyBuf) -> Result, io::Error> { - match buf.as_slice().iter().position(|&b| b == b'\n') { - Some(i) => Ok(Some(buf.drain_to(i + 1).into())), + fn decode(&mut self, buf: &mut BytesMut) -> Result, io::Error> { + match buf.iter().position(|&b| b == b'\n') { + Some(i) => Ok(Some(buf.split_to(i + 1).into())), None => Ok(None), } } - fn decode_eof(&mut self, buf: &mut EasyBuf) -> io::Result { - let amt = buf.len(); - Ok(buf.drain_to(amt)) + fn decode_eof(&mut self, buf: &mut BytesMut) -> io::Result> { + if buf.len() == 0 { + Ok(None) + } else { + let amt = buf.len(); + Ok(Some(buf.split_to(amt))) + } } +} - fn encode(&mut self, item: EasyBuf, into: &mut Vec) -> io::Result<()> { - into.extend_from_slice(item.as_slice()); +impl Encoder for LineCodec { + type Item = BytesMut; + type Error = io::Error; + + fn encode(&mut self, item: BytesMut, into: &mut BytesMut) -> io::Result<()> { + into.put(&item[..]); Ok(()) } } diff --git a/tests/pipe-hup.rs b/tests/pipe-hup.rs index 9fae3e760..8ea1095a4 100644 --- a/tests/pipe-hup.rs +++ b/tests/pipe-hup.rs @@ -5,6 +5,7 @@ extern crate futures; extern crate libc; extern crate mio; extern crate tokio_core; +extern crate tokio_io; use std::fs::File; use std::io::{self, Write}; @@ -12,11 +13,11 @@ use std::os::unix::io::{AsRawFd, FromRawFd}; use std::thread; use std::time::Duration; -use mio::{Evented, PollOpt, Ready, Token}; -use mio::unix::EventedFd; - -use tokio_core::io::read_to_end; +use mio::unix::{UnixReady, EventedFd}; +use mio::{PollOpt, Ready, Token}; +use mio::event::Evented; use tokio_core::reactor::{Core, PollEvented}; +use tokio_io::io::read_to_end; macro_rules! t { ($e:expr) => (match $e { @@ -46,11 +47,13 @@ impl io::Read for MyFile { impl Evented for MyFile { fn register(&self, poll: &mio::Poll, token: Token, interest: Ready, opts: PollOpt) -> io::Result<()> { - EventedFd(&self.0.as_raw_fd()).register(poll, token, interest | Ready::hup(), opts) + let hup: Ready = UnixReady::hup().into(); + EventedFd(&self.0.as_raw_fd()).register(poll, token, interest | hup, opts) } fn reregister(&self, poll: &mio::Poll, token: Token, interest: Ready, opts: PollOpt) -> io::Result<()> { - EventedFd(&self.0.as_raw_fd()).reregister(poll, token, interest | Ready::hup(), opts) + let hup: Ready = UnixReady::hup().into(); + EventedFd(&self.0.as_raw_fd()).reregister(poll, token, interest | hup, opts) } fn deregister(&self, poll: &mio::Poll) -> io::Result<()> { EventedFd(&self.0.as_raw_fd()).deregister(poll) diff --git a/tests/spawn.rs b/tests/spawn.rs index 8fd4599d5..11732f0d5 100644 --- a/tests/spawn.rs +++ b/tests/spawn.rs @@ -6,6 +6,8 @@ use std::time::Duration; use std::sync::mpsc; use futures::Future; +use futures::future; +use futures::sync::oneshot; use tokio_core::reactor::Core; #[test] @@ -13,15 +15,15 @@ fn simple() { drop(env_logger::init()); let mut lp = Core::new().unwrap(); - let (tx1, rx1) = futures::oneshot(); - let (tx2, rx2) = futures::oneshot(); - lp.handle().spawn(futures::lazy(|| { - tx1.complete(1); + let (tx1, rx1) = oneshot::channel(); + let (tx2, rx2) = oneshot::channel(); + lp.handle().spawn(future::lazy(|| { + tx1.send(1).unwrap(); Ok(()) })); lp.remote().spawn(|_| { - futures::lazy(|| { - tx2.complete(2); + future::lazy(|| { + tx2.send(2).unwrap(); Ok(()) }) }); @@ -38,12 +40,12 @@ fn simple_core_poll() { let (tx1, tx2) = (tx.clone(), tx.clone()); lp.turn(Some(Duration::new(0, 0))); - lp.handle().spawn(futures::lazy(move || { + lp.handle().spawn(future::lazy(move || { tx1.send(1).unwrap(); Ok(()) })); lp.turn(Some(Duration::new(0, 0))); - lp.handle().spawn(futures::lazy(move || { + lp.handle().spawn(future::lazy(move || { tx2.send(2).unwrap(); Ok(()) })); @@ -58,14 +60,14 @@ fn spawn_in_poll() { drop(env_logger::init()); let mut lp = Core::new().unwrap(); - let (tx1, rx1) = futures::oneshot(); - let (tx2, rx2) = futures::oneshot(); + let (tx1, rx1) = oneshot::channel(); + let (tx2, rx2) = oneshot::channel(); let remote = lp.remote(); - lp.handle().spawn(futures::lazy(move || { - tx1.complete(1); + lp.handle().spawn(future::lazy(move || { + tx1.send(1).unwrap(); remote.spawn(|_| { - futures::lazy(|| { - tx2.complete(2); + future::lazy(|| { + tx2.send(2).unwrap(); Ok(()) }) }); diff --git a/tests/stream-buffered.rs b/tests/stream-buffered.rs index 262b4f7ad..012fd0540 100644 --- a/tests/stream-buffered.rs +++ b/tests/stream-buffered.rs @@ -1,6 +1,7 @@ +extern crate env_logger; extern crate futures; extern crate tokio_core; -extern crate env_logger; +extern crate tokio_io; use std::io::{Read, Write}; use std::net::TcpStream; @@ -8,7 +9,8 @@ use std::thread; use futures::Future; use futures::stream::Stream; -use tokio_core::io::{Io, copy}; +use tokio_io::io::copy; +use tokio_io::AsyncRead; use tokio_core::net::TcpListener; use tokio_core::reactor::Core;