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

Don't store an Arc in ReadinessStream

Browse Source 
This commit contains a few refactorings, but the major goal is to remove the
`Arc` that's stored inside of each `ReadinessStream` and `Scheduled` slot in the
event loop. The original purpose of this `Arc` was to share the I/O object among
the concrete handle itself and the event loop. The event loop would then change
how the socket is registered over time and then deregister it when it gets a
"shutdown request".

Nowadays, however, once an I/O object is registered with the event loop it's
never updated. Additionally, we don't actually need to call `deregister` but can
rather just instead close the I/O object itself and let the kernel/event loop
take care of the cleanup. All we need to do on deregistering is free up the slab
entry.

The major result of this commit is that I/O objects no longer need to be `Sync`
(as they're not stored in an `Arc`). Instead they just need to be `Send +
'static` as one might otherwise expect.

Along the way this also refactors a few pieces here and there to make more sense
in this new scheme. The `ReadinessStream` type now has a type parameter
indicating an owned reference to the I/O object it wraps. This can be accessed
via the `get_ref` and `get_mut` methods. Additionally I/O tokens on the event
loop are now a full-fledged `IoToken` type which we can change in the future if
we need to.
This commit is contained in:
Alex Crichton
2016-08-20 23:07:00 -07:00
parent 5ab323e5c2
commit b9dae23e3f
5 changed files with 226 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/event_loop.rs
+102 -100
View File
@@ -80,7 +80,7 @@ pub struct LoopPin {
}
struct Scheduled {
source: IoSource,
readiness: Arc<AtomicUsize>,
reader: Option<TaskHandle>,
writer: Option<TaskHandle>,
}
@@ -97,7 +97,6 @@ enum Direction {
}
enum Message {
AddSource(IoSource, Arc<Slot<io::Result<usize>>>),
DropSource(usize),
Schedule(usize, TaskHandle, Direction),
AddTimeout(Instant, Arc<Slot<io::Result<TimeoutToken>>>),
@@ -107,29 +106,6 @@ enum Message {
Drop(DropBox<dropbox::MyDrop>),
}
/// Type of I/O objects inserted into the event loop, created by `Source::new`.
pub struct Source<E: ?Sized> {
readiness: AtomicUsize,
io: E,
}
/// I/O objects inserted into the event loop
pub type IoSource = Arc<Source<mio::Evented + Sync + Send>>;
fn register(poll: &mio::Poll,
token: usize,
sched: &Scheduled) -> io::Result<()> {
poll.register(&sched.source.io,
mio::Token(token),
mio::EventSet::readable() | mio::EventSet::writable(),
mio::PollOpt::edge())
}
fn deregister(poll: &mio::Poll, sched: &Scheduled) {
// TODO: handle error
poll.deregister(&sched.source.io).unwrap();
}
impl Loop {
/// Creates a new event loop, returning any error that happened during the
/// creation.
@@ -334,11 +310,11 @@ impl Loop {
if let Some(sched) = self.dispatch.borrow_mut().get_mut(token) {
if event.kind().is_readable() {
reader = sched.reader.take();
sched.source.readiness.fetch_or(1, Ordering::Relaxed);
sched.readiness.fetch_or(1, Ordering::Relaxed);
}
if event.kind().is_writable() {
writer = sched.writer.take();
sched.source.readiness.fetch_or(2, Ordering::Relaxed);
sched.readiness.fetch_or(2, Ordering::Relaxed);
}
} else {
debug!("notified on {} which no longer exists", token);
@@ -382,10 +358,10 @@ impl Loop {
CURRENT_LOOP.set(&self, || handle.unpark());
}
fn add_source(&self, source: IoSource) -> io::Result<usize> {
fn add_source(&self, source: &mio::Evented) -> io::Result<IoToken> {
debug!("adding a new I/O source");
let sched = Scheduled {
source: source,
readiness: Arc::new(AtomicUsize::new(0)),
reader: None,
writer: None,
};
@@ -395,14 +371,20 @@ impl Loop {
dispatch.grow(amt);
}
let entry = dispatch.vacant_entry().unwrap();
try!(register(&self.io, entry.index(), &sched));
Ok(entry.insert(sched).index())
try!(self.io.register(source,
mio::Token(entry.index()),
mio::EventSet::readable() |
mio::EventSet::writable(),
mio::PollOpt::edge()));
Ok(IoToken {
readiness: sched.readiness.clone(),
token: entry.insert(sched).index(),
})
}
fn drop_source(&self, token: usize) {
debug!("dropping I/O source: {}", token);
let sched = self.dispatch.borrow_mut().remove(token).unwrap();
deregister(&self.io, &sched);
self.dispatch.borrow_mut().remove(token).unwrap();
}
fn schedule(&self, token: usize, wake: TaskHandle, dir: Direction) {
@@ -414,10 +396,10 @@ impl Loop {
Direction::Read => (&mut sched.reader, 1),
Direction::Write => (&mut sched.writer, 2),
};
let ready = sched.source.readiness.load(Ordering::SeqCst);
let ready = sched.readiness.load(Ordering::SeqCst);
if ready & bit != 0 {
*slot = None;
sched.source.readiness.store(ready & !bit, Ordering::SeqCst);
sched.readiness.store(ready & !bit, Ordering::SeqCst);
Some(wake)
} else {
*slot = Some(wake);
@@ -472,11 +454,6 @@ impl Loop {
fn notify(&self, msg: Message) {
match msg {
Message::AddSource(source, slot) => {
// This unwrap() should always be ok as we're the only producer
slot.try_produce(self.add_source(source))
.ok().expect("interference with try_produce");
}
Message::DropSource(tok) => self.drop_source(tok),
Message::Schedule(tok, wake, dir) => self.schedule(tok, wake, dir),
@@ -545,19 +522,20 @@ impl LoopHandle {
///
/// When a new I/O object is created it needs to be communicated to the
/// event loop to ensure that it's registered and ready to receive
/// notifications. The event loop with then respond with a unique token that
/// this handle can be identified with (the resolved value of the returned
/// future).
/// notifications. The event loop with then respond back with the I/O object
/// and a token which can be used to send more messages to the event loop.
///
/// This token is then passed in turn to each of the methods below to
/// interact with notifications on the I/O object itself.
/// The token returned is then passed in turn to each of the methods below
/// to interact with notifications on the I/O object itself.
///
/// # Panics
///
/// 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 add_source(&self, source: IoSource) -> AddSource {
pub fn add_source<E>(&self, source: E) -> AddSource<E>
where E: mio::Evented + Send + 'static,
{
AddSource {
inner: LoopFuture {
loop_handle: self.clone(),
@@ -567,15 +545,19 @@ impl LoopHandle {
}
}
/// Begin listening for read events on an event loop.
/// Schedule the current future task to receive a notification when the
/// corresponding I/O object is readable.
///
/// Once an I/O object has been registered with the event loop through the
/// `add_source` method, this method can be used with the assigned token to
/// begin awaiting read notifications.
/// notify the current future task when the next read notification comes in.
///
/// Currently the current task will be notified with *edge* semantics. This
/// means that whenever the underlying I/O object changes state, e.g. it was
/// not readable and now it is, then a notification will be sent.
/// The current task will only receive a notification **once** and to
/// receive further notifications it will need to call `schedule_read`
/// again.
///
/// > **Note**: This method should generally not be used directly, but
/// > rather the `ReadinessStream` type should be used instead.
///
/// # Panics
///
@@ -585,19 +567,24 @@ impl LoopHandle {
///
/// This function will also panic if there is not a currently running future
/// task.
pub fn schedule_read(&self, tok: usize) {
self.send(Message::Schedule(tok, task::park(), Direction::Read));
pub fn schedule_read(&self, tok: &IoToken) {
self.send(Message::Schedule(tok.token, task::park(), Direction::Read));
}
/// Begin listening for write events on an event loop.
/// Schedule the current future task to receive a notification when the
/// corresponding I/O object is writable.
///
/// Once an I/O object has been registered with the event loop through the
/// `add_source` method, this method can be used with the assigned token to
/// begin awaiting write notifications.
/// notify the current future task when the next write notification comes
/// in.
///
/// Currently the current task will be notified with *edge* semantics. This
/// means that whenever the underlying I/O object changes state, e.g. it was
/// not writable and now it is, then a notification will be sent.
/// The current task will only receive a notification **once** and to
/// receive further notifications it will need to call `schedule_write`
/// again.
///
/// > **Note**: This method should generally not be used directly, but
/// > rather the `ReadinessStream` type should be used instead.
///
/// # Panics
///
@@ -607,8 +594,8 @@ impl LoopHandle {
///
/// This function will also panic if there is not a currently running future
/// task.
pub fn schedule_write(&self, tok: usize) {
self.send(Message::Schedule(tok, task::park(), Direction::Write));
pub fn schedule_write(&self, tok: &IoToken) {
self.send(Message::Schedule(tok.token, task::park(), Direction::Write));
}
/// Unregister all information associated with a token on an event loop,
@@ -625,13 +612,16 @@ impl LoopHandle {
/// ensure that the callbacks are **not** invoked, so pending scheduled
/// callbacks cannot be relied upon to get called.
///
/// > **Note**: This method should generally not be used directly, but
/// > rather the `ReadinessStream` type should be used instead.
///
/// # Panics
///
/// This function 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 drop_source(&self, tok: usize) {
self.send(Message::DropSource(tok));
pub fn drop_source(&self, tok: &IoToken) {
self.send(Message::DropSource(tok.token));
}
/// Adds a new timeout to get fired at the specified instant, notifying the
@@ -731,16 +721,60 @@ impl LoopPin {
///
/// Created through the `LoopHandle::add_source` method, this future can also
/// resolve to an error if there's an issue communicating with the event loop.
pub struct AddSource {
inner: LoopFuture<usize, IoSource>,
pub struct AddSource<E> {
inner: LoopFuture<(E, IoToken), E>,
}
impl Future for AddSource {
type Item = usize;
/// A token that identifies an active timeout.
pub struct IoToken {
token: usize,
// TODO: can we avoid this allocation? It's kind of a bummer...
readiness: Arc<AtomicUsize>,
}
impl IoToken {
/// Consumes the last readiness notification the token this source is for
/// registered.
///
/// Currently sources receive readiness notifications on an edge-basis. That
/// is, once you receive a notification that an object can be read, you
/// won't receive any more notifications until all of that data has been
/// read.
///
/// The event loop will fill in this information and then inform futures
/// that they're ready to go with the `schedule` method, and then the `poll`
/// method can use this to figure out what happened.
///
/// > **Note**: This method should generally not be used directly, but
/// > rather the `ReadinessStream` type should be used instead.
// TODO: this should really return a proper newtype/enum, not a usize
pub fn take_readiness(&self) -> usize {
self.readiness.swap(0, Ordering::SeqCst)
}
}
impl<E> Future for AddSource<E>
where E: mio::Evented + Send + 'static,
{
type Item = (E, IoToken);
type Error = io::Error;
fn poll(&mut self) -> Poll<usize, io::Error> {
self.inner.poll(Loop::add_source, Message::AddSource)
fn poll(&mut self) -> Poll<(E, IoToken), io::Error> {
let handle = self.inner.loop_handle.clone();
self.inner.poll(|lp, io| {
let token = try!(lp.add_source(&io));
Ok((io, token))
}, |io, slot| {
Message::Run(Box::new(move || {
let res = handle.with_loop(|lp| {
let lp = lp.unwrap();
let token = try!(lp.add_source(&io));
Ok((io, token))
});
slot.try_produce(res).ok()
.expect("add source try_produce intereference");
}))
})
}
}
@@ -1114,38 +1148,6 @@ impl TimeoutState {
}
}
impl<E> Source<E> {
/// Creates a new `Source` wrapping the provided source of events.
pub fn new(e: E) -> Source<E> {
Source {
readiness: AtomicUsize::new(0),
io: e,
}
}
}
impl<E: ?Sized> Source<E> {
/// Consumes the last readiness notification that this source received.
///
/// Currently sources receive readiness notifications on an edge-basis. That
/// is, once you receive a notification that an object can be read, you
/// won't receive any more notifications until all of that data has been
/// read.
///
/// The event loop will fill in this information and then inform futures
/// that they're ready to go with the `schedule` method, and then the `poll`
/// method can use this to figure out what happened.
// TODO: shouldn't return a usize here, but rather some kind of newtype
pub fn take_readiness(&self) -> usize {
self.readiness.swap(0, Ordering::SeqCst)
}
/// Gets access to the underlying I/O object.
pub fn io(&self) -> &E {
&self.io
}
}
impl Executor for MioSender {
fn execute_boxed(&self, callback: Box<ExecuteCallback>) {
self.inner.send(Message::Run(callback))
src/lib.rs
+1 -1
View File
@@ -30,7 +30,7 @@ mod mpsc_queue;
mod channel;
pub use event_loop::{Loop, LoopPin, LoopHandle, AddSource, AddTimeout};
pub use event_loop::{LoopData, AddLoopData, TimeoutToken, IoSource, Source};
pub use event_loop::{LoopData, AddLoopData, TimeoutToken, IoToken};
pub use readiness_stream::ReadinessStream;
pub use tcp::{TcpListener, TcpStream};
pub use timeout::Timeout;
src/readiness_stream.rs
+50 -30
View File
@@ -2,8 +2,9 @@ use std::io;
use std::sync::atomic::{AtomicUsize, Ordering};
use futures::{Future, Poll};
use mio;
use event_loop::{IoSource, LoopHandle, AddSource};
use event_loop::{IoToken, LoopHandle, AddSource};
/// A concrete implementation of a stream of readiness notifications for I/O
/// objects that originates from an event loop.
@@ -21,31 +22,30 @@ use event_loop::{IoSource, LoopHandle, AddSource};
/// It's the responsibility of the wrapper to inform the readiness stream when a
/// "would block" I/O event is seen. The readiness stream will then take care of
/// any scheduling necessary to get notified when the event is ready again.
pub struct ReadinessStream {
io_token: usize,
loop_handle: LoopHandle,
source: IoSource,
pub struct ReadinessStream<E> {
token: IoToken,
handle: LoopHandle,
readiness: AtomicUsize,
io: E,
}
pub struct ReadinessStreamNew {
inner: AddSource,
handle: Option<LoopHandle>,
source: Option<IoSource>,
pub struct ReadinessStreamNew<E> {
inner: AddSource<E>,
handle: LoopHandle,
}
impl ReadinessStream {
impl<E> ReadinessStream<E>
where E: mio::Evented + Send + 'static,
{
/// Creates a new readiness stream associated with the provided
/// `loop_handle` and for the given `source`.
///
/// This method returns a future which will resolve to the readiness stream
/// when it's ready.
pub fn new(loop_handle: LoopHandle, source: IoSource)
-> ReadinessStreamNew {
pub fn new(loop_handle: LoopHandle, source: E) -> ReadinessStreamNew<E> {
ReadinessStreamNew {
inner: loop_handle.add_source(source.clone()),
source: Some(source),
handle: Some(loop_handle),
inner: loop_handle.add_source(source),
handle: loop_handle,
}
}
@@ -60,11 +60,11 @@ impl ReadinessStream {
if self.readiness.load(Ordering::SeqCst) & 1 != 0 {
return Poll::Ok(())
}
self.readiness.fetch_or(self.source.take_readiness(), Ordering::SeqCst);
self.readiness.fetch_or(self.token.take_readiness(), Ordering::SeqCst);
if self.readiness.load(Ordering::SeqCst) & 1 != 0 {
Poll::Ok(())
} else {
self.loop_handle.schedule_read(self.io_token);
self.handle.schedule_read(&self.token);
Poll::NotReady
}
}
@@ -80,11 +80,11 @@ impl ReadinessStream {
if self.readiness.load(Ordering::SeqCst) & 2 != 0 {
return Poll::Ok(())
}
self.readiness.fetch_or(self.source.take_readiness(), Ordering::SeqCst);
self.readiness.fetch_or(self.token.take_readiness(), Ordering::SeqCst);
if self.readiness.load(Ordering::SeqCst) & 2 != 0 {
Poll::Ok(())
} else {
self.loop_handle.schedule_write(self.io_token);
self.handle.schedule_write(&self.token);
Poll::NotReady
}
}
@@ -102,7 +102,7 @@ impl ReadinessStream {
/// then again readable.
pub fn need_read(&self) {
self.readiness.fetch_and(!1, Ordering::SeqCst);
self.loop_handle.schedule_read(self.io_token);
self.handle.schedule_read(&self.token);
}
/// Indicates to this source of events that the corresponding I/O object is
@@ -118,28 +118,48 @@ impl ReadinessStream {
/// then again writable.
pub fn need_write(&self) {
self.readiness.fetch_and(!2, Ordering::SeqCst);
self.loop_handle.schedule_write(self.io_token);
self.handle.schedule_write(&self.token);
}
/// Returns a reference to the event loop handle that this readiness stream
/// is associated with.
pub fn loop_handle(&self) -> &LoopHandle {
&self.handle
}
/// Returns a shared reference to the underlying I/O object this readiness
/// stream is wrapping.
pub fn get_ref(&self) -> &E {
&self.io
}
/// Returns a mutable reference to the underlying I/O object this readiness
/// stream is wrapping.
pub fn get_mut(&mut self) -> &mut E {
&mut self.io
}
}
impl Future for ReadinessStreamNew {
type Item = ReadinessStream;
impl<E> Future for ReadinessStreamNew<E>
where E: mio::Evented + Send + 'static,
{
type Item = ReadinessStream<E>;
type Error = io::Error;
fn poll(&mut self) -> Poll<ReadinessStream, io::Error> {
self.inner.poll().map(|token| {
fn poll(&mut self) -> Poll<ReadinessStream<E>, io::Error> {
self.inner.poll().map(|(io, token)| {
ReadinessStream {
io_token: token,
source: self.source.take().unwrap(),
loop_handle: self.handle.take().unwrap(),
token: token,
handle: self.handle.clone(),
io: io,
readiness: AtomicUsize::new(0),
}
})
}
}
impl Drop for ReadinessStream {
impl<E> Drop for ReadinessStream<E> {
fn drop(&mut self) {
self.loop_handle.drop_source(self.io_token)
self.handle.drop_source(&self.token)
}
}
src/tcp.rs
+46 -63
View File
@@ -2,7 +2,6 @@ use std::fmt;
use std::io::{self, ErrorKind, Read, Write};
use std::mem;
use std::net::{self, SocketAddr, Shutdown};
use std::sync::Arc;
use futures::stream::Stream;
use futures::{Future, IntoFuture, failed, Poll};
@@ -10,27 +9,21 @@ use futures_io::{IoFuture, IoStream};
use mio;
use {ReadinessStream, LoopHandle};
use event_loop::Source;
/// An I/O object representing a TCP socket listening for incoming connections.
///
/// This object can be converted into a stream of incoming connections for
/// various forms of processing.
pub struct TcpListener {
loop_handle: LoopHandle,
ready: ReadinessStream,
listener: Arc<Source<mio::tcp::TcpListener>>,
io: ReadinessStream<mio::tcp::TcpListener>,
}
impl TcpListener {
fn new(listener: mio::tcp::TcpListener,
handle: LoopHandle) -> IoFuture<TcpListener> {
let listener = Arc::new(Source::new(listener));
ReadinessStream::new(handle.clone(), listener.clone()).map(|r| {
ReadinessStream::new(handle, listener).map(|io| {
TcpListener {
loop_handle: handle,
ready: r,
listener: listener,
io: io,
}
}).boxed()
}
@@ -73,7 +66,7 @@ impl TcpListener {
/// Test whether this socket is ready to be read or not.
pub fn poll_read(&self) -> Poll<(), io::Error> {
self.ready.poll_read()
self.io.poll_read()
}
/// Returns the local address that this listener is bound to.
@@ -81,7 +74,7 @@ impl TcpListener {
/// This can be useful, for example, when binding to port 0 to figure out
/// which port was actually bound.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.listener.io().local_addr()
self.io.get_ref().local_addr()
}
/// Consumes this listener, returning a stream of the sockets this listener
@@ -99,14 +92,14 @@ impl TcpListener {
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, io::Error> {
match self.inner.listener.io().accept() {
match self.inner.io.get_ref().accept() {
Ok(Some(pair)) => {
debug!("accepted a socket");
Poll::Ok(Some(pair))
}
Ok(None) => {
debug!("waiting to accept another socket");
self.inner.ready.need_read();
self.inner.io.need_read();
Poll::NotReady
}
Err(e) => Poll::Err(e),
@@ -114,17 +107,11 @@ impl TcpListener {
}
}
let loop_handle = self.loop_handle.clone();
let loop_handle = self.io.loop_handle().clone();
Incoming { inner: self }
.and_then(move |(tcp, addr)| {
let tcp = Arc::new(Source::new(tcp));
ReadinessStream::new(loop_handle.clone(),
tcp.clone()).map(move |ready| {
let stream = TcpStream {
source: tcp,
ready: ready,
};
(stream, addr)
ReadinessStream::new(loop_handle.clone(), tcp).map(move |io| {
(TcpStream { io: io }, addr)
})
}).boxed()
}
@@ -132,7 +119,7 @@ impl TcpListener {
impl fmt::Debug for TcpListener {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
self.listener.io().fmt(f)
self.io.get_ref().fmt(f)
}
}
@@ -143,8 +130,7 @@ impl fmt::Debug for TcpListener {
/// raw underlying I/O object as well as streams for the read/write
/// notifications on the stream itself.
pub struct TcpStream {
source: Arc<Source<mio::tcp::TcpStream>>,
ready: ReadinessStream,
io: ReadinessStream<mio::tcp::TcpStream>,
}
enum TcpStreamNew {
@@ -184,18 +170,8 @@ impl TcpStream {
fn new(connected_stream: mio::tcp::TcpStream,
handle: LoopHandle)
-> IoFuture<TcpStream> {
// Once we've connected, wait for the stream to be writable as that's
// when the actual connection has been initiated. Once we're writable we
// check for `take_socket_error` to see if the connect actually hit an
// error or not.
//
// If all that succeeded then we ship everything on up.
let connected_stream = Arc::new(Source::new(connected_stream));
ReadinessStream::new(handle, connected_stream.clone()).and_then(|ready| {
TcpStreamNew::Waiting(TcpStream {
source: connected_stream,
ready: ready,
})
ReadinessStream::new(handle, connected_stream).and_then(|io| {
TcpStreamNew::Waiting(TcpStream { io: io })
}).boxed()
}
@@ -233,7 +209,7 @@ impl TcpStream {
/// is only suitable for calling in a `Future::poll` method and will
/// automatically handle ensuring a retry once the socket is readable again.
pub fn poll_read(&self) -> Poll<(), io::Error> {
self.ready.poll_read()
self.io.poll_read()
}
/// Test whether this socket is writey to be written to or not.
@@ -243,17 +219,17 @@ impl TcpStream {
/// is only suitable for calling in a `Future::poll` method and will
/// automatically handle ensuring a retry once the socket is writable again.
pub fn poll_write(&self) -> Poll<(), io::Error> {
self.ready.poll_write()
self.io.poll_write()
}
/// Returns the local address that this stream is bound to.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.source.io().local_addr()
self.io.get_ref().local_addr()
}
/// Returns the remote address that this stream is connected to.
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
self.source.io().peer_addr()
self.io.get_ref().peer_addr()
}
/// Shuts down the read, write, or both halves of this connection.
@@ -262,7 +238,7 @@ impl TcpStream {
/// portions to return immediately with an appropriate value (see the
/// documentation of `Shutdown`).
pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
self.source.io().shutdown(how)
self.io.get_ref().shutdown(how)
}
/// Sets the value of the `TCP_NODELAY` option on this socket.
@@ -273,12 +249,12 @@ impl TcpStream {
/// sufficient amount to send out, thereby avoiding the frequent sending of
/// small packets.
pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
self.source.io().set_nodelay(nodelay)
self.io.get_ref().set_nodelay(nodelay)
}
/// Sets the keepalive time in seconds for this socket.
pub fn set_keepalive_s(&self, seconds: Option<u32>) -> io::Result<()> {
self.source.io().set_keepalive(seconds)
self.io.get_ref().set_keepalive(seconds)
}
}
@@ -291,9 +267,16 @@ impl Future for TcpStreamNew {
TcpStreamNew::Waiting(s) => s,
TcpStreamNew::Empty => panic!("can't poll TCP stream twice"),
};
match stream.ready.poll_write() {
// Once we've connected, wait for the stream to be writable as that's
// when the actual connection has been initiated. Once we're writable we
// check for `take_socket_error` to see if the connect actually hit an
// error or not.
//
// If all that succeeded then we ship everything on up.
match stream.io.poll_write() {
Poll::Ok(()) => {
match stream.source.io().take_socket_error() {
match stream.io.get_ref().take_socket_error() {
Ok(()) => return Poll::Ok(stream),
Err(ref e) if e.kind() == ErrorKind::WouldBlock => {}
Err(e) => return Poll::Err(e),
@@ -309,9 +292,9 @@ impl Future for TcpStreamNew {
impl Read for TcpStream {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let r = self.source.io().read(buf);
let r = self.io.get_ref().read(buf);
if is_wouldblock(&r) {
self.ready.need_read();
self.io.need_read();
}
return r
}
@@ -319,16 +302,16 @@ impl Read for TcpStream {
impl Write for TcpStream {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let r = self.source.io().write(buf);
let r = self.io.get_ref().write(buf);
if is_wouldblock(&r) {
self.ready.need_write();
self.io.need_write();
}
return r
}
fn flush(&mut self) -> io::Result<()> {
let r = self.source.io().flush();
let r = self.io.get_ref().flush();
if is_wouldblock(&r) {
self.ready.need_write();
self.io.need_write();
}
return r
}
@@ -336,9 +319,9 @@ impl Write for TcpStream {
impl<'a> Read for &'a TcpStream {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let r = self.source.io().read(buf);
let r = self.io.get_ref().read(buf);
if is_wouldblock(&r) {
self.ready.need_read();
self.io.need_read();
}
return r
}
@@ -346,17 +329,17 @@ impl<'a> Read for &'a TcpStream {
impl<'a> Write for &'a TcpStream {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let r = self.source.io().write(buf);
let r = self.io.get_ref().write(buf);
if is_wouldblock(&r) {
self.ready.need_write();
self.io.need_write();
}
return r
}
fn flush(&mut self) -> io::Result<()> {
let r = self.source.io().flush();
let r = self.io.get_ref().flush();
if is_wouldblock(&r) {
self.ready.need_write();
self.io.need_write();
}
return r
}
@@ -371,7 +354,7 @@ fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
impl fmt::Debug for TcpStream {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
self.source.io().fmt(f)
self.io.get_ref().fmt(f)
}
}
@@ -382,13 +365,13 @@ mod sys {
impl AsRawFd for TcpStream {
fn as_raw_fd(&self) -> RawFd {
self.source.io().as_raw_fd()
self.io.get_ref().as_raw_fd()
}
}
impl AsRawFd for TcpListener {
fn as_raw_fd(&self) -> RawFd {
self.listener.io().as_raw_fd()
self.io.get_ref().as_raw_fd()
}
}
}
@@ -402,7 +385,7 @@ mod sys {
//
// impl AsRawHandle for TcpStream {
// fn as_raw_handle(&self) -> RawHandle {
// self.source.io().as_raw_handle()
// self.io.get_ref().as_raw_handle()
// }
// }
//
src/udp.rs
+27 -34
View File
@@ -1,6 +1,5 @@
use std::io;
use std::net::{self, SocketAddr, Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
use std::fmt;
use futures::{Future, failed, Poll};
@@ -8,12 +7,10 @@ use futures_io::IoFuture;
use mio;
use {ReadinessStream, LoopHandle};
use event_loop::Source;
/// An I/O object representing a UDP socket.
pub struct UdpSocket {
source: Arc<Source<mio::udp::UdpSocket>>,
ready: ReadinessStream,
io: ReadinessStream<mio::udp::UdpSocket>,
}
impl LoopHandle {
@@ -34,12 +31,8 @@ impl LoopHandle {
impl UdpSocket {
fn new(socket: mio::udp::UdpSocket, handle: LoopHandle)
-> IoFuture<UdpSocket> {
let socket = Arc::new(Source::new(socket));
ReadinessStream::new(handle, socket.clone()).map(|ready| {
UdpSocket {
source: socket,
ready: ready,
}
ReadinessStream::new(handle, socket).map(|io| {
UdpSocket { io: io }
}).boxed()
}
@@ -62,7 +55,7 @@ impl UdpSocket {
/// Returns the local address that this stream is bound to.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.source.io().local_addr()
self.io.get_ref().local_addr()
}
/// Test whether this socket is ready to be read or not.
@@ -72,7 +65,7 @@ impl UdpSocket {
/// is only suitable for calling in a `Future::poll` method and will
/// automatically handle ensuring a retry once the socket is readable again.
pub fn poll_read(&self) -> Poll<(), io::Error> {
self.ready.poll_read()
self.io.poll_read()
}
/// Test whether this socket is writey to be written to or not.
@@ -82,7 +75,7 @@ impl UdpSocket {
/// is only suitable for calling in a `Future::poll` method and will
/// automatically handle ensuring a retry once the socket is writable again.
pub fn poll_write(&self) -> Poll<(), io::Error> {
self.ready.poll_write()
self.io.poll_write()
}
/// Sends data on the socket to the given address. On success, returns the
@@ -91,10 +84,10 @@ impl UdpSocket {
/// Address type can be any implementor of `ToSocketAddrs` trait. See its
/// documentation for concrete examples.
pub fn send_to(&self, buf: &[u8], target: &SocketAddr) -> io::Result<usize> {
match self.source.io().send_to(buf, target) {
match self.io.get_ref().send_to(buf, target) {
Ok(Some(n)) => Ok(n),
Ok(None) => {
self.ready.need_write();
self.io.need_write();
Err(io::Error::new(io::ErrorKind::WouldBlock, "would block"))
}
Err(e) => Err(e),
@@ -104,10 +97,10 @@ impl UdpSocket {
/// Receives data from the socket. On success, returns the number of bytes
/// read and the address from whence the data came.
pub fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
match self.source.io().recv_from(buf) {
match self.io.get_ref().recv_from(buf) {
Ok(Some(n)) => Ok(n),
Ok(None) => {
self.ready.need_read();
self.io.need_read();
Err(io::Error::new(io::ErrorKind::WouldBlock, "would block"))
}
Err(e) => Err(e),
@@ -121,7 +114,7 @@ impl UdpSocket {
///
/// [link]: #method.set_broadcast
pub fn broadcast(&self) -> io::Result<bool> {
self.source.io().broadcast()
self.io.get_ref().broadcast()
}
/// Sets the value of the `SO_BROADCAST` option for this socket.
@@ -129,7 +122,7 @@ impl UdpSocket {
/// When enabled, this socket is allowed to send packets to a broadcast
/// address.
pub fn set_broadcast(&self, on: bool) -> io::Result<()> {
self.source.io().set_broadcast(on)
self.io.get_ref().set_broadcast(on)
}
/// Gets the value of the `IP_MULTICAST_LOOP` option for this socket.
@@ -139,7 +132,7 @@ impl UdpSocket {
///
/// [link]: #method.set_multicast_loop_v4
pub fn multicast_loop_v4(&self) -> io::Result<bool> {
self.source.io().multicast_loop_v4()
self.io.get_ref().multicast_loop_v4()
}
/// Sets the value of the `IP_MULTICAST_LOOP` option for this socket.
@@ -147,7 +140,7 @@ impl UdpSocket {
/// If enabled, multicast packets will be looped back to the local socket.
/// Note that this may not have any affect on IPv6 sockets.
pub fn set_multicast_loop_v4(&self, on: bool) -> io::Result<()> {
self.source.io().set_multicast_loop_v4(on)
self.io.get_ref().set_multicast_loop_v4(on)
}
/// Gets the value of the `IP_MULTICAST_TTL` option for this socket.
@@ -157,7 +150,7 @@ impl UdpSocket {
///
/// [link]: #method.set_multicast_ttl_v4
pub fn multicast_ttl_v4(&self) -> io::Result<u32> {
self.source.io().multicast_ttl_v4()
self.io.get_ref().multicast_ttl_v4()
}
/// Sets the value of the `IP_MULTICAST_TTL` option for this socket.
@@ -168,7 +161,7 @@ impl UdpSocket {
///
/// Note that this may not have any affect on IPv6 sockets.
pub fn set_multicast_ttl_v4(&self, ttl: u32) -> io::Result<()> {
self.source.io().set_multicast_ttl_v4(ttl)
self.io.get_ref().set_multicast_ttl_v4(ttl)
}
/// Gets the value of the `IPV6_MULTICAST_LOOP` option for this socket.
@@ -178,7 +171,7 @@ impl UdpSocket {
///
/// [link]: #method.set_multicast_loop_v6
pub fn multicast_loop_v6(&self) -> io::Result<bool> {
self.source.io().multicast_loop_v6()
self.io.get_ref().multicast_loop_v6()
}
/// Sets the value of the `IPV6_MULTICAST_LOOP` option for this socket.
@@ -186,7 +179,7 @@ impl UdpSocket {
/// Controls whether this socket sees the multicast packets it sends itself.
/// Note that this may not have any affect on IPv4 sockets.
pub fn set_multicast_loop_v6(&self, on: bool) -> io::Result<()> {
self.source.io().set_multicast_loop_v6(on)
self.io.get_ref().set_multicast_loop_v6(on)
}
/// Gets the value of the `IP_TTL` option for this socket.
@@ -195,7 +188,7 @@ impl UdpSocket {
///
/// [link]: #method.set_ttl
pub fn ttl(&self) -> io::Result<u32> {
self.source.io().ttl()
self.io.get_ref().ttl()
}
/// Sets the value for the `IP_TTL` option on this socket.
@@ -203,7 +196,7 @@ impl UdpSocket {
/// This value sets the time-to-live field that is used in every packet sent
/// from this socket.
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
self.source.io().set_ttl(ttl)
self.io.get_ref().set_ttl(ttl)
}
/// Executes an operation of the `IP_ADD_MEMBERSHIP` type.
@@ -216,7 +209,7 @@ impl UdpSocket {
pub fn join_multicast_v4(&self,
multiaddr: &Ipv4Addr,
interface: &Ipv4Addr) -> io::Result<()> {
self.source.io().join_multicast_v4(multiaddr, interface)
self.io.get_ref().join_multicast_v4(multiaddr, interface)
}
/// Executes an operation of the `IPV6_ADD_MEMBERSHIP` type.
@@ -227,7 +220,7 @@ impl UdpSocket {
pub fn join_multicast_v6(&self,
multiaddr: &Ipv6Addr,
interface: u32) -> io::Result<()> {
self.source.io().join_multicast_v6(multiaddr, interface)
self.io.get_ref().join_multicast_v6(multiaddr, interface)
}
/// Executes an operation of the `IP_DROP_MEMBERSHIP` type.
@@ -239,7 +232,7 @@ impl UdpSocket {
pub fn leave_multicast_v4(&self,
multiaddr: &Ipv4Addr,
interface: &Ipv4Addr) -> io::Result<()> {
self.source.io().leave_multicast_v4(multiaddr, interface)
self.io.get_ref().leave_multicast_v4(multiaddr, interface)
}
/// Executes an operation of the `IPV6_DROP_MEMBERSHIP` type.
@@ -251,13 +244,13 @@ impl UdpSocket {
pub fn leave_multicast_v6(&self,
multiaddr: &Ipv6Addr,
interface: u32) -> io::Result<()> {
self.source.io().leave_multicast_v6(multiaddr, interface)
self.io.get_ref().leave_multicast_v6(multiaddr, interface)
}
}
impl fmt::Debug for UdpSocket {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
self.source.io().fmt(f)
self.io.get_ref().fmt(f)
}
}
@@ -268,7 +261,7 @@ mod sys {
impl AsRawFd for UdpSocket {
fn as_raw_fd(&self) -> RawFd {
self.source.io().as_raw_fd()
self.io.get_ref().as_raw_fd()
}
}
}
@@ -282,7 +275,7 @@ mod sys {
//
// impl AsRawHandle for UdpSocket {
// fn as_raw_handle(&self) -> RawHandle {
// self.source.io().as_raw_handle()
// self.io.get_ref().as_raw_handle()
// }
// }
}