From bc64194be17bc9711b4a56364e677d823c7cc3d1 Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Sat, 30 Jul 2016 17:53:12 -0700 Subject: [PATCH] Let's rename everything! --- Cargo.toml | 18 ++ src/bin/echo.rs | 53 +++++ src/bin/sink.rs | 51 +++++ src/buf_reader.rs | 174 ++++++++++++++ src/buf_writer.rs | 179 +++++++++++++++ src/event_loop.rs | 489 ++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 46 ++++ src/readiness_stream.rs | 85 +++++++ src/tcp.rs | 324 ++++++++++++++++++++++++++ tests/buffered.rs | 64 ++++++ tests/chain.rs | 52 +++++ tests/echo.rs | 48 ++++ tests/limit.rs | 44 ++++ tests/tcp.rs | 81 +++++++ 14 files changed, 1708 insertions(+) create mode 100644 Cargo.toml create mode 100644 src/bin/echo.rs create mode 100644 src/bin/sink.rs create mode 100755 src/buf_reader.rs create mode 100755 src/buf_writer.rs create mode 100644 src/event_loop.rs create mode 100644 src/lib.rs create mode 100755 src/readiness_stream.rs create mode 100644 src/tcp.rs create mode 100644 tests/buffered.rs create mode 100644 tests/chain.rs create mode 100644 tests/echo.rs create mode 100644 tests/limit.rs create mode 100644 tests/tcp.rs diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 000000000..7783e50a8 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "futures-mio" +version = "0.1.0" +authors = ["Alex Crichton "] + +[dependencies] +futures = { path = ".." } +futures-io = { path = "../futures-io" } +log = "0.3" +mio = { git = "https://github.com/alexcrichton/mio", branch = "write-then-drop" } +scoped-tls = "0.1.0" +slab = "0.2.0" + +[dev-dependencies] +env_logger = "0.3" + +[lib] +test = false diff --git a/src/bin/echo.rs b/src/bin/echo.rs new file mode 100644 index 000000000..c7826f2f4 --- /dev/null +++ b/src/bin/echo.rs @@ -0,0 +1,53 @@ +//! An echo server that just writes back everything that's written to it. + +extern crate futures; +extern crate futures_io; +extern crate futures_mio; + +use std::env; +use std::net::SocketAddr; + +use futures::Future; +use futures_io::{copy, TaskIo}; +use futures::stream::Stream; + +fn main() { + let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = addr.parse::().unwrap(); + + // Create the event loop that will drive this server + let mut l = futures_mio::Loop::new().unwrap(); + + // Create a TCP listener which will listen for incoming connections + let server = l.handle().tcp_listen(&addr); + + let done = server.and_then(move |socket| { + // Once we've got the TCP listener, inform that we have it + println!("Listenering on: {}", addr); + + // Pull out the stream of incoming connections and then for each new + // one spin up a new task copying data. We put the `socket` into a + // `TaskIo` structure which then allows us to `split` it into the read + // and write halves of the socket. + // + // Finally we use the `io::copy` future to copy all data from the + // reading half onto the writing half. + socket.incoming().for_each(|(socket, addr)| { + let io = TaskIo::new(socket); + let pair = io.map(|io| io.split()); + let amt = pair.and_then(|(reader, writer)| { + copy(reader, writer) + }); + + // Once all that is done we print out how much we wrote, and then + // critically we *forget* this future which allows it to run + // concurrently with other connections. + amt.map(move |amt| { + println!("wrote {} bytes to {}", amt, addr) + }).forget(); + + Ok(()) + }) + }); + l.run(done).unwrap(); +} diff --git a/src/bin/sink.rs b/src/bin/sink.rs new file mode 100644 index 000000000..3ee6743ff --- /dev/null +++ b/src/bin/sink.rs @@ -0,0 +1,51 @@ +//! A small server that writes as many nul bytes on all connections it receives. +//! +//! There is no concurrency in this server, only one connection is written to at +//! a time. + +#[macro_use] +extern crate futures; +extern crate futures_mio; + +use std::env; +use std::io::{self, Write}; +use std::net::SocketAddr; + +use futures::Future; +use futures::stream::Stream; + +fn main() { + let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string()); + let addr = addr.parse::().unwrap(); + + let mut l = futures_mio::Loop::new().unwrap(); + let server = l.handle().tcp_listen(&addr).and_then(|socket| { + socket.incoming().and_then(|(socket, addr)| { + println!("got a socket: {}", addr); + write(socket) + }).for_each(|()| { + println!("lost the socket"); + Ok(()) + }) + }); + println!("Listenering on: {}", addr); + l.run(server).unwrap(); +} + +fn write(socket: futures_mio::TcpStream) -> Box> { + static BUF: &'static [u8] = &[0; 64 * 1024]; + socket.into_future().map_err(|e| e.0).and_then(move |(ready, mut socket)| { + let ready = match ready { + Some(ready) => ready, + None => return futures::finished(()).boxed(), + }; + while ready.is_write() { + match socket.write(&BUF) { + Ok(_) => {} + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break, + Err(e) => return futures::failed(e).boxed(), + } + } + write(socket) + }).boxed() +} diff --git a/src/buf_reader.rs b/src/buf_reader.rs new file mode 100755 index 000000000..66fad5dee --- /dev/null +++ b/src/buf_reader.rs @@ -0,0 +1,174 @@ +#![allow(missing_docs)] + +use std::io::{self, Read}; +use std::ops::Deref; +use std::sync::Arc; + +use ReadinessStream; + +use futures::{Task, Poll}; +use futures::stream::Stream; + +const INPUT_BUF_SIZE: usize = 8 * 1024; + +/// A cheap to copy, read-only slice of an input buffer. +#[derive(Clone)] +pub struct InputBuf { + buf: Arc>, + pos: usize, + len: usize, +} + +impl Deref for InputBuf { + type Target = [u8]; + + fn deref(&self) -> &[u8] { + &self.buf[self.pos..self.pos + self.len] + } +} + +// TODO: implement direct slicing (which clones the Arc) + +impl InputBuf { + fn new() -> InputBuf { + InputBuf { + buf: Arc::new(Vec::with_capacity(INPUT_BUF_SIZE)), + pos: 0, + len: 0, + } + } + + pub fn take(&mut self, len: usize) -> InputBuf { + assert!(len <= self.len); + let new = InputBuf { + buf: self.buf.clone(), + pos: self.pos, + len: len, + }; + self.pos += len; + self.len -= len; + new + } + + pub fn skip(&mut self, len: usize) { + assert!(len <= self.len); + self.pos += len; + } + + fn with_mut(&mut self, f: F) -> R + where F: FnOnce(&mut Vec) -> R + { + // Fast path if we can get mutable access to our own current + // buffer. + if let Some(buf) = Arc::get_mut(&mut self.buf) { + buf.drain(..self.pos); + self.pos = 0; + let ret = f(buf); + self.len = buf.len(); + return ret; + } + + // If we couldn't get access above then we give ourself a new buffer + // here. + + let mut v = Vec::with_capacity(INPUT_BUF_SIZE); + v.extend_from_slice(&self.buf[self.pos..]); + let ret = f(&mut v); + + self.buf = Arc::new(v); + self.pos = 0; + self.len = self.buf.len(); + ret + } + + fn read(&mut self, socket: &mut R) -> io::Result<(usize, bool)> { + unsafe fn slice_to_end(v: &mut Vec) -> &mut [u8] { + use std::slice; + if v.capacity() == 0 { + v.reserve(16); + } + if v.capacity() == v.len() { + v.reserve(1); + } + slice::from_raw_parts_mut(v.as_mut_ptr().offset(v.len() as isize), + v.capacity() - v.len()) + } + + self.with_mut(|buf| { + match socket.read(unsafe { slice_to_end(buf) }) { + Ok(0) => { + trace!("socket EOF"); + Ok((0, true)) + } + Ok(n) => { + trace!("socket read {} bytes", n); + unsafe { + let len = buf.len(); + buf.set_len(len + n); + } + Ok((n, false)) + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Ok((0, false)), + Err(e) => Err(e), + } + }) + } +} + +/// A stream for parsing from an underlying reader, using an unbounded internal +/// buffer. +pub struct BufReader { + source: R, + source_ready: ReadinessStream, + read_ready: bool, + buf: InputBuf, +} + +impl BufReader { + pub fn new(source: R, source_ready: ReadinessStream) -> BufReader { + BufReader { + source: source, + source_ready: source_ready, + read_ready: false, + buf: InputBuf::new(), + } + } + + pub fn buf(&mut self) -> &mut InputBuf { + &mut self.buf + } +} + +impl Stream for BufReader { + type Item = (); + type Error = io::Error; + + fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { + if !self.read_ready { + match self.source_ready.poll(task) { + Poll::NotReady => return Poll::NotReady, + Poll::Err(e) => return Poll::Err(e.into()), + Poll::Ok(Some(ref r)) if !r.is_read() => return Poll::NotReady, + Poll::Ok(Some(_)) => self.read_ready = true, + _ => unreachable!(), + } + } + + match self.buf.read(&mut self.source) { + Ok((0, true)) => Poll::Ok(None), + Ok((0, false)) => { + self.read_ready = false; + Poll::NotReady + } + Ok(_) => { + self.read_ready = true; + Poll::Ok(Some(())) + } + Err(e) => Poll::Err(e.into()), + } + } + + fn schedule(&mut self, task: &mut Task) { + self.source_ready.schedule(task) + } +} diff --git a/src/buf_writer.rs b/src/buf_writer.rs new file mode 100755 index 000000000..2e87db98d --- /dev/null +++ b/src/buf_writer.rs @@ -0,0 +1,179 @@ +#![allow(missing_docs)] + +use std::io::{self, Write}; + +use futures::{Future, Task, Poll}; +use futures::stream::Stream; +use ReadinessStream; + +const OUTPUT_BUF_SIZE: usize = 8 * 1024; + +pub struct BufWriter { + sink: W, + sink_ready: ReadinessStream, + write_ready: bool, + buf: Vec, +} + +impl BufWriter { + pub fn new(sink: W, sink_ready: ReadinessStream) -> BufWriter { + BufWriter { + sink: sink, + sink_ready: sink_ready, + write_ready: false, + buf: Vec::with_capacity(OUTPUT_BUF_SIZE), + } + } + + pub fn extend(&mut self, data: &[u8]) { + extend(&mut self.buf, data) + } + + pub fn flush(self) -> Flush { + Flush { writer: Some(self) } + } + + pub fn reserve(self, amt: usize) -> Reserve { + Reserve { amt: amt, writer: Some(self) } + } + + /// Is there buffered data waiting to be sent? + pub fn is_dirty(&self) -> bool { + self.buf.len() > 0 + } + + fn poll_flush(&mut self, task: &mut Task) -> Poll<(), io::Error> { + let mut task = task.scoped(); + while self.is_dirty() { + if !self.write_ready { + match self.sink_ready.poll(&mut task) { + Poll::Err(e) => return Poll::Err(e), + Poll::Ok(Some(ref r)) if !r.is_write() => return Poll::NotReady, + Poll::Ok(Some(_)) => self.write_ready = true, + Poll::Ok(None) | // TODO: this should translate to an error + Poll::NotReady => return Poll::NotReady, + } + } + + debug!("trying to write some data"); + match self.sink.write(&self.buf) { + Ok(0) => return Poll::Err(io::Error::new(io::ErrorKind::Other, "early eof")), + Ok(n) => { + // TODO: consider draining more lazily, i.e. only just + // before returning + self.buf.drain(..n); + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.write_ready = false; + } + Err(e) => return Poll::Err(e), + } + + task.ready(); + } + + debug!("fully flushed"); + Poll::Ok(()) + } +} + +impl Write for BufWriter { + fn write(&mut self, data: &[u8]) -> io::Result { + extend(&mut self.buf, data); + Ok(data.len()) + } + + fn flush(&mut self) -> io::Result<()> { + // TODO: something reasonable + unimplemented!() + } +} + +pub struct Flush { + writer: Option>, +} + +impl Flush { + pub fn is_dirty(&self) -> bool { + self.writer.as_ref().unwrap().is_dirty() + } + + pub fn into_inner(mut self) -> BufWriter { + self.writer.take().unwrap() + } +} + +impl Future for Flush { + type Item = BufWriter; + type Error = (io::Error, BufWriter); + + fn poll(&mut self, task: &mut Task) + -> Poll, (io::Error, BufWriter)> { + match self.writer.as_mut().unwrap().poll_flush(task) { + Poll::Ok(()) => Poll::Ok(self.writer.take().unwrap()), + Poll::Err(e) => Poll::Err((e, self.writer.take().unwrap())), + Poll::NotReady => Poll::NotReady, + } + } + + fn schedule(&mut self, task: &mut Task) { + let writer = self.writer.as_mut().unwrap(); + + assert!(!writer.write_ready); + writer.sink_ready.schedule(task) + } +} + +// TODO: why doesn't extend_from_slice optimize to this? +fn extend(dst: &mut Vec, data: &[u8]) { + use std::ptr; + dst.reserve(data.len()); + let prev = dst.len(); + unsafe { + ptr::copy_nonoverlapping(data.as_ptr(), + dst.as_mut_ptr().offset(prev as isize), + data.len()); + dst.set_len(prev + data.len()); + } +} + +pub struct Reserve { + amt: usize, + writer: Option>, +} + +impl Future for Reserve { + type Item = BufWriter; + type Error = (io::Error, BufWriter); + + fn poll(&mut self, task: &mut Task) + -> Poll, (io::Error, BufWriter)> { + loop { + let (cap, len) = { + let buf = &mut self.writer.as_mut().unwrap().buf; + (buf.capacity(), buf.len()) + }; + + if self.amt <= cap - len { + return Poll::Ok(self.writer.take().unwrap()) + } else if self.amt > cap { + let mut writer = self.writer.take().unwrap(); + writer.buf.reserve(self.amt); + return Poll::Ok(writer) + } + + match self.writer.as_mut().unwrap().poll_flush(task) { + Poll::Ok(()) => {}, + Poll::Err(e) => return Poll::Err((e, self.writer.take().unwrap())), + Poll::NotReady => return Poll::NotReady, + } + } + } + + fn schedule(&mut self, task: &mut Task) { + let writer = self.writer.as_mut().unwrap(); + + assert!(!writer.write_ready); + writer.sink_ready.schedule(task) + } +} diff --git a/src/event_loop.rs b/src/event_loop.rs new file mode 100644 index 000000000..dbb4eb7a5 --- /dev/null +++ b/src/event_loop.rs @@ -0,0 +1,489 @@ +use std::cell::{Cell, RefCell}; +use std::io::{self, ErrorKind}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering}; +use std::sync::mpsc; +use std::time::Instant; + +use mio; +use mio::channel::SendError; +use slab::Slab; +use futures::{Future, Task, TaskHandle, Poll}; +use futures_io::Ready; + +use slot::{self, Slot}; + +static NEXT_LOOP_ID: AtomicUsize = ATOMIC_USIZE_INIT; +scoped_thread_local!(static CURRENT_LOOP: Loop); + +const SLAB_CAPACITY: usize = 1024 * 64; + +/// An event loop. +/// +/// The event loop is the main source of blocking in an application which drives +/// all other I/O events and notifications happening. Each event loop can have +/// multiple handles pointing to it, each of which can then be used to create +/// various I/O objects to interact with the event loop in interesting ways. +// TODO: expand this +pub struct Loop { + id: usize, + active: Cell, + io: mio::Poll, + tx: mio::channel::Sender, + rx: mio::channel::Receiver, + dispatch: RefCell>, +} + +/// Handle to an event loop, used to construct I/O objects, send messages, and +/// otherwise interact indirectly with the event loop itself. +/// +/// Handles can be cloned, and when cloned they will still refer to the +/// same underlying event loop. +#[derive(Clone)] +pub struct LoopHandle { + id: usize, + tx: mio::channel::Sender, +} + +struct Scheduled { + source: IoSource, + waiter: Option, +} + +enum Message { + AddSource(IoSource, Arc>>), + DropSource(usize), + Schedule(usize, TaskHandle), + Deschedule(usize), + Shutdown, +} + +pub struct Source { + readiness: AtomicUsize, + io: E, +} + +pub type IoSource = Arc>; + +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. + pub fn new() -> io::Result { + let (tx, rx) = mio::channel::from_std_channel(mpsc::channel()); + let io = try!(mio::Poll::new()); + try!(io.register(&rx, + mio::Token(0), + mio::EventSet::readable(), + mio::PollOpt::edge())); + Ok(Loop { + id: NEXT_LOOP_ID.fetch_add(1, Ordering::Relaxed), + active: Cell::new(true), + io: io, + tx: tx, + rx: rx, + dispatch: RefCell::new(Slab::new_starting_at(1, SLAB_CAPACITY)), + }) + } + + /// Generates a handle to this event loop used to construct I/O objects and + /// send messages. + /// + /// Handles to an event loop are cloneable as well and clones will always + /// refer to the same event loop. + pub fn handle(&self) -> LoopHandle { + LoopHandle { + id: self.id, + tx: self.tx.clone(), + } + } + + #[allow(missing_docs)] + pub fn run(&mut self, f: F) -> Result { + let (tx_res, rx_res) = mpsc::channel(); + let handle = self.handle(); + f.then(move |res| { + handle.shutdown(); + tx_res.send(res) + }).forget(); + + self._run(); + + rx_res.recv().unwrap() + } + + fn _run(&mut self) { + let mut events = mio::Events::new(); + self.active.set(true); + while self.active.get() { + let amt; + // On Linux, Poll::poll is epoll_wait, which may return EINTR if a + // ptracer attaches. This retry loop prevents crashing when + // attaching strace, or similar. + let start = Instant::now(); + loop { + match self.io.poll(&mut events, None) { + Ok(a) => { + amt = a; + break; + } + Err(ref e) if e.kind() == ErrorKind::Interrupted => {} + err @ Err(_) => { + err.unwrap(); + } + } + } + debug!("loop poll - {:?}", start.elapsed()); + + // TODO: coalesce token sets for a given Wake? + let start = Instant::now(); + for i in 0..events.len() { + let event = events.get(i).unwrap(); + let token = usize::from(event.token()); + + if token == 0 { + debug!("consuming notification queue"); + self.consume_queue(); + continue + } + + let mut waiter = None; + + if let Some(sched) = self.dispatch.borrow_mut().get_mut(token) { + waiter = sched.waiter.take(); + if event.kind().is_readable() { + sched.source.readiness.fetch_or(1, Ordering::Relaxed); + } + if event.kind().is_writable() { + sched.source.readiness.fetch_or(2, Ordering::Relaxed); + } + } else { + debug!("notified on {} which no longer exists", token); + } + debug!("dispatching {:?} {:?}", event.token(), event.kind()); + + CURRENT_LOOP.set(&self, move || { + match waiter { + Some(waiter) => waiter.notify(), + None => debug!("no waiter"), + } + }); + } + + debug!("loop process - {} events, {:?}", amt, start.elapsed()); + } + + debug!("loop is done!"); + } + + fn add_source(&self, source: IoSource) -> io::Result { + let sched = Scheduled { + source: source, + waiter: None, + }; + let mut dispatch = self.dispatch.borrow_mut(); + if dispatch.vacant_entry().is_none() { + let amt = dispatch.count(); + dispatch.grow(amt); + } + let entry = dispatch.vacant_entry().unwrap(); + try!(register(&self.io, entry.index(), &sched)); + Ok(entry.insert(sched).index()) + } + + fn drop_source(&self, token: usize) { + let sched = self.dispatch.borrow_mut().remove(token).unwrap(); + deregister(&self.io, &sched); + } + + fn schedule(&self, token: usize, wake: TaskHandle) { + let to_call = { + let mut dispatch = self.dispatch.borrow_mut(); + let sched = dispatch.get_mut(token).unwrap(); + if sched.source.readiness.load(Ordering::Relaxed) != 0 { + sched.waiter = None; + Some(wake) + } else { + sched.waiter = Some(wake); + None + } + }; + if let Some(to_call) = to_call { + to_call.notify(); + } + } + + fn deschedule(&self, token: usize) { + let mut dispatch = self.dispatch.borrow_mut(); + dispatch.get_mut(token).unwrap(); + } + + fn consume_queue(&self) { + while let Ok(msg) = self.rx.try_recv() { + self.notify(msg); + } + } + + 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) => self.schedule(tok, wake), + Message::Deschedule(tok) => self.deschedule(tok), + Message::Shutdown => self.active.set(false), + } + } +} + +impl LoopHandle { + fn send(&self, msg: Message) { + self.with_loop(|lp| { + match lp { + Some(lp) => { + // Need to execute all existing requests first, to ensure + // that our message is processed "in order" + lp.consume_queue(); + lp.notify(msg); + } + None => { + match self.tx.send(msg) { + Ok(()) => {} + + // This should only happen when there was an error + // writing to the pipe to wake up the event loop, + // hopefully that never happens + Err(SendError::Io(e)) => { + panic!("error sending message to event loop: {}", e) + } + + // If we're still sending a message to the event loop + // after it's closed, then that's bad! + Err(SendError::Disconnected(_)) => { + panic!("event loop is no longer available") + } + } + } + } + }) + } + + fn with_loop(&self, f: F) -> R + where F: FnOnce(Option<&Loop>) -> R + { + if CURRENT_LOOP.is_set() { + CURRENT_LOOP.with(|lp| { + if lp.id == self.id { + f(Some(lp)) + } else { + f(None) + } + }) + } else { + f(None) + } + } + + /// Add a new source to an event loop, returning a future which will resolve + /// to the token that can be used to identify this source. + /// + /// 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). + /// + /// This token 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 { + AddSource { + loop_handle: self.clone(), + source: Some(source), + result: None, + } + } + + fn add_source_(&self, source: IoSource, slot: Arc>>) { + self.send(Message::AddSource(source, slot)); + } + + /// Begin listening for events on an event loop. + /// + /// 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 notifications. + /// + /// The `dir` argument indicates how the I/O object is expected to be + /// awaited on (either readable or writable) and the `wake` callback will be + /// invoked. Note that one the `wake` callback is invoked once it will not + /// be invoked again, it must be re-`schedule`d to continue receiving + /// notifications. + /// + /// # 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 schedule(&self, tok: usize, task: &mut Task) { + // TODO: plumb through `&mut Task` if we're on the event loop + self.send(Message::Schedule(tok, task.handle().clone())); + } + + /// Stop listening for events on an event loop. + /// + /// Once a callback has been scheduled with the `schedule` method, it can be + /// unregistered from the event loop with this method. This method does not + /// guarantee that the callback will not be invoked if it hasn't already, + /// but a best effort will be made to ensure it is not called. + /// + /// # 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 deschedule(&self, tok: usize) { + self.send(Message::Deschedule(tok)); + } + + /// Unregister all information associated with a token on an event loop, + /// deallocating all internal resources assigned to the given token. + /// + /// This method should be called whenever a source of events is being + /// destroyed. This will ensure that the event loop can reuse `tok` for + /// another I/O object if necessary and also remove it from any poll + /// notifications and callbacks. + /// + /// Note that wake callbacks may still be invoked after this method is + /// called as it may take some time for the message to drop a source to + /// reach the event loop. Despite this fact, this method will attempt to + /// ensure that the callbacks are **not** invoked, so pending scheduled + /// callbacks cannot be relied upon to get called. + /// + /// # 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)); + } + + /// Send a message to the associated event loop that it should shut down, or + /// otherwise break out of its current loop of iteration. + /// + /// This method does not forcibly cause the event loop to shut down or + /// perform an interrupt on whatever task is currently running, instead a + /// message is simply enqueued to at a later date process the request to + /// stop looping ASAP. + /// + /// # 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 shutdown(&self) { + self.send(Message::Shutdown); + } +} + +/// A future which will resolve a unique `tok` token for an I/O object. +/// +/// 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 { + loop_handle: LoopHandle, + source: Option, + result: Option<(Arc>>, slot::Token)>, +} + +impl Future for AddSource { + type Item = usize; + type Error = io::Error; + + fn poll(&mut self, _task: &mut Task) -> Poll { + match self.result { + Some((ref result, ref token)) => { + result.cancel(*token); + match result.try_consume() { + Ok(t) => t.into(), + Err(_) => Poll::NotReady, + } + } + None => { + let source = &mut self.source; + self.loop_handle.with_loop(|lp| { + match lp { + Some(lp) => lp.add_source(source.take().unwrap()).into(), + None => Poll::NotReady, + } + }) + } + } + } + + fn schedule(&mut self, task: &mut Task) { + if let Some((ref result, ref mut token)) = self.result { + result.cancel(*token); + let handle = task.handle().clone(); + *token = result.on_full(move |_| { + handle.notify(); + }); + return + } + + let handle = task.handle().clone(); + let result = Arc::new(Slot::new(None)); + let token = result.on_full(move |_| { + handle.notify(); + }); + self.result = Some((result.clone(), token)); + self.loop_handle.add_source_(self.source.take().unwrap(), result); + } +} + +impl Source { + pub fn new(e: E) -> Source { + Source { + readiness: AtomicUsize::new(0), + io: e, + } + } +} + +impl Source { + pub fn take_readiness(&self) -> Option { + match self.readiness.swap(0, Ordering::SeqCst) { + 0 => None, + 1 => Some(Ready::Read), + 2 => Some(Ready::Write), + 3 => Some(Ready::ReadWrite), + _ => panic!(), + } + } + + pub fn io(&self) -> &E { + &self.io + } +} diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 000000000..cefe37d9c --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,46 @@ +//! A binding to mio giving it a future/stream interface on top. +//! +//! This library contains the rudimentary bindings to an event loop in mio which +//! provides future and stream-based abstractions of all the underlying I/O +//! objects that mio provides internally. +//! +//! Currently very much a work in progress, and breakage should be expected! + +#![deny(missing_docs)] + +extern crate futures; +extern crate futures_io; +extern crate mio; +extern crate slab; + +#[macro_use] +extern crate scoped_tls; + +#[macro_use] +extern crate log; + +use std::io; +use futures::Future; +use futures::stream::Stream; + +mod readiness_stream; +mod event_loop; +mod tcp; +mod buf_reader; +mod buf_writer; +#[path = "../../src/slot.rs"] +mod slot; +#[path = "../../src/lock.rs"] +mod lock; + +/// A convenience typedef around a `Future` whose error component is `io::Error` +pub type IoFuture = Future; + +/// A convenience typedef around a `Stream` whose error component is `io::Error` +pub type IoStream = Stream; + +pub use event_loop::{Loop, LoopHandle}; +pub use readiness_stream::ReadinessStream; +pub use tcp::{TcpListener, TcpStream}; +pub use buf_reader::{BufReader, InputBuf}; +pub use buf_writer::{BufWriter, Flush, Reserve}; diff --git a/src/readiness_stream.rs b/src/readiness_stream.rs new file mode 100755 index 000000000..b89868208 --- /dev/null +++ b/src/readiness_stream.rs @@ -0,0 +1,85 @@ +#![allow(missing_docs)] // TODO: document this module + +use std::io; +use std::sync::Arc; + +use futures::stream::Stream; +use futures::{Future, Task, Poll}; +use futures_io::Ready; + +use IoFuture; +use event_loop::{IoSource, LoopHandle}; +use readiness_stream::drop_source::DropSource; + + +// TODO: figure out a nicer way to factor this +mod drop_source { + use event_loop::LoopHandle; + + pub struct DropSource { + token: usize, + loop_handle: LoopHandle, + } + + impl DropSource { + pub fn new(token: usize, loop_handle: LoopHandle) -> DropSource { + DropSource { + token: token, + loop_handle: loop_handle, + } + } + } + + // Safe because no public access exposed to LoopHandle; only used in drop + unsafe impl Sync for DropSource {} + + impl Drop for DropSource { + fn drop(&mut self) { + self.loop_handle.drop_source(self.token) + } + } +} + +pub struct ReadinessStream { + io_token: usize, + loop_handle: LoopHandle, + source: IoSource, + _drop_source: Arc, +} + +impl ReadinessStream { + pub fn new(loop_handle: LoopHandle, source: IoSource) + -> Box> { + loop_handle.add_source(source.clone()).map(|token| { + let drop_source = Arc::new(DropSource::new(token, loop_handle.clone())); + ReadinessStream { + io_token: token, + source: source, + loop_handle: loop_handle, + _drop_source: drop_source, + } + }).boxed() + } +} + +impl Stream for ReadinessStream { + type Item = Ready; + type Error = io::Error; + + fn poll(&mut self, _task: &mut Task) -> Poll, io::Error> { + match self.source.take_readiness() { + None => Poll::NotReady, + Some(r) => Poll::Ok(Some(r)), + } + } + + fn schedule(&mut self, task: &mut Task) { + self.loop_handle.schedule(self.io_token, task) + } +} + +impl Drop for ReadinessStream { + fn drop(&mut self) { + self.loop_handle.deschedule(self.io_token) + } +} diff --git a/src/tcp.rs b/src/tcp.rs new file mode 100644 index 000000000..3aec07723 --- /dev/null +++ b/src/tcp.rs @@ -0,0 +1,324 @@ +use std::io::{self, ErrorKind, Read, Write}; +use std::mem; +use std::net::{self, SocketAddr}; +use std::sync::Arc; + +use futures::stream::{self, Stream}; +use futures::{Future, IntoFuture, failed, Task, Poll}; +use futures_io::Ready; +use mio; + +use {IoFuture, IoStream, 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>, +} + +impl TcpListener { + fn new(listener: mio::tcp::TcpListener, + handle: LoopHandle) -> Box> { + let listener = Arc::new(Source::new(listener)); + ReadinessStream::new(handle.clone(), listener.clone()).map(|r| { + TcpListener { + loop_handle: handle, + ready: r, + listener: listener, + } + }).boxed() + } + + /// Create a new TCP listener from the standard library's TCP listener. + /// + /// This method can be used when the `LoopHandle::tcp_listen` method isn't + /// sufficient because perhaps some more configuration is needed in terms of + /// before the calls to `bind` and `listen`. + /// + /// This API is typically paired with the `net2` crate and the `TcpBuilder` + /// type to build up and customize a listener before it's shipped off to the + /// backing event loop. This allows configuration of options like + /// `SO_REUSEPORT`, binding to multiple addresses, etc. + /// + /// The `addr` argument here is one of the addresses that `listener` is + /// bound to and the listener will only be guaranteed to accept connections + /// of the same address type currently. + /// + /// Finally, the `handle` argument is the event loop that this listener will + /// be bound to. + /// + /// The platform specific behavior of this function looks like: + /// + /// * On Unix, the socket is placed into nonblocking mode and connections + /// can be accepted as normal + /// + /// * On Windows, the address is stored internally and all future accepts + /// will only be for the same IP version as `addr` specified. That is, if + /// `addr` is an IPv4 address then all sockets accepted will be IPv4 as + /// well (same for IPv6). + pub fn from_listener(listener: net::TcpListener, + addr: &SocketAddr, + handle: LoopHandle) -> Box> { + mio::tcp::TcpListener::from_listener(listener, addr) + .into_future() + .and_then(|l| TcpListener::new(l, handle)) + .boxed() + } + + /// Returns the local address that this listener is bound to. + /// + /// 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 { + self.listener.io().local_addr() + } + + /// Consumes this listener, returning a stream of the sockets this listener + /// accepts. + /// + /// This method returns an implementation of the `Stream` trait which + /// resolves to the sockets the are accepted on this listener. + pub fn incoming(self) -> Box> { + let TcpListener { loop_handle, listener, ready } = self; + + ready + .map(move |_| { + stream::iter(NonblockingIter { source: listener.clone() }.fuse()) + }) + .flatten() + .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) + }) + }).boxed() + } +} + +struct NonblockingIter { + source: Arc>, +} + +impl Iterator for NonblockingIter { + type Item = io::Result<(mio::tcp::TcpStream, SocketAddr)>; + + fn next(&mut self) -> Option> { + match self.source.io().accept() { + Ok(Some(e)) => { + debug!("accepted connection"); + Some(Ok(e)) + } + Ok(None) => { + debug!("no connection ready"); + None + } + Err(e) => Some(Err(e)), + } + } +} + +impl Stream for TcpListener { + type Item = Ready; + type Error = io::Error; + + fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { + self.ready.poll(task) + } + + fn schedule(&mut self, task: &mut Task) { + self.ready.schedule(task) + } +} + +/// An I/O object representing a TCP stream connected to a remote endpoint. +/// +/// A TCP stream can either be created by connecting to an endpoint or by +/// accepting a connection from a listener. Inside the stream is access to the +/// raw underlying I/O object as well as streams for the read/write +/// notifications on the stream itself. +pub struct TcpStream { + source: Arc>, + ready: ReadinessStream, +} + +enum TcpStreamNew { + Waiting(TcpStream), + Empty, +} + +impl LoopHandle { + /// Create a new TCP listener associated with this event loop. + /// + /// The TCP listener will bind to the provided `addr` address, if available, + /// and will be returned as a future. The returned future, if resolved + /// successfully, can then be used to accept incoming connections. + pub fn tcp_listen(self, addr: &SocketAddr) -> Box> { + match mio::tcp::TcpListener::bind(addr) { + Ok(l) => TcpListener::new(l, self), + Err(e) => failed(e).boxed(), + } + } + + /// Create a new TCP stream connected to the specified address. + /// + /// This function will create a new TCP socket and attempt to connect it to + /// the `addr` provided. The returned future will be resolved once the + /// stream has successfully connected. If an error happens during the + /// connection or during the socket creation, that error will be returned to + /// the future instead. + pub fn tcp_connect(self, addr: &SocketAddr) -> Box> { + match mio::tcp::TcpStream::connect(addr) { + Ok(tcp) => TcpStream::new(tcp, self), + Err(e) => failed(e).boxed(), + } + } +} + +impl TcpStream { + fn new(connected_stream: mio::tcp::TcpStream, + handle: LoopHandle) + -> Box> { + // 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, + }) + }).boxed() + } + + /// Creates a new `TcpStream` from the pending socket inside the given + /// `std::net::TcpStream`, connecting it to the address specified. + /// + /// This constructor allows configuring the socket before it's actually + /// connected, and this function will transfer ownership to the returned + /// `TcpStream` if successful. An unconnected `TcpStream` can be created + /// with the `net2::TcpBuilder` type (and also configured via that route). + /// + /// The platform specific behavior of this function looks like: + /// + /// * On Unix, the socket is placed into nonblocking mode and then a + /// `connect` call is issued. + /// + /// * On Windows, the address is stored internally and the connect operation + /// is issued when the returned `TcpStream` is registered with an event + /// loop. Note that on Windows you must `bind` a socket before it can be + /// connected, so if a custom `TcpBuilder` is used it should be bound + /// (perhaps to `INADDR_ANY`) before this method is called. + pub fn connect_stream(stream: net::TcpStream, + addr: &SocketAddr, + handle: LoopHandle) -> Box> { + match mio::tcp::TcpStream::connect_stream(stream, addr) { + Ok(tcp) => TcpStream::new(tcp, handle), + Err(e) => failed(e).boxed(), + } + } + + /// Returns the local address that this stream is bound to. + pub fn local_addr(&self) -> io::Result { + self.source.io().local_addr() + } + + /// Returns the remote address that this stream is connected to. + pub fn peer_addr(&self) -> io::Result { + self.source.io().peer_addr() + } +} + +impl Future for TcpStreamNew { + type Item = TcpStream; + type Error = io::Error; + + fn poll(&mut self, task: &mut Task) -> Poll { + let mut stream = match mem::replace(self, TcpStreamNew::Empty) { + TcpStreamNew::Waiting(s) => s, + TcpStreamNew::Empty => panic!("can't poll TCP stream twice"), + }; + match stream.ready.poll(task) { + Poll::Ok(None) => panic!(), + Poll::Ok(Some(_)) => { + match stream.source.io().take_socket_error() { + Ok(()) => return Poll::Ok(stream), + Err(ref e) if e.kind() == ErrorKind::WouldBlock => {} + Err(e) => return Poll::Err(e), + } + } + Poll::Err(e) => return Poll::Err(e), + Poll::NotReady => {} + } + *self = TcpStreamNew::Waiting(stream); + Poll::NotReady + } + + fn schedule(&mut self, task: &mut Task) { + match *self { + TcpStreamNew::Waiting(ref mut s) => { + s.ready.schedule(task); + } + TcpStreamNew::Empty => task.notify(), + } + } +} + +impl Read for TcpStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + let r = self.source.io().read(buf); + trace!("read[{:p}] {:?} on {:?}", self, r, self.source.io()); + return r + } +} + +impl Write for TcpStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + let r = self.source.io().write(buf); + trace!("write[{:p}] {:?} on {:?}", self, r, self.source.io()); + return r + } + fn flush(&mut self) -> io::Result<()> { + self.source.io().flush() + } +} + +impl<'a> Read for &'a TcpStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.source.io().read(buf) + } +} + +impl<'a> Write for &'a TcpStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.source.io().write(buf) + } + fn flush(&mut self) -> io::Result<()> { + self.source.io().flush() + } +} + +impl Stream for TcpStream { + type Item = Ready; + type Error = io::Error; + + fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { + self.ready.poll(task) + } + + fn schedule(&mut self, task: &mut Task) { + self.ready.schedule(task) + } +} diff --git a/tests/buffered.rs b/tests/buffered.rs new file mode 100644 index 000000000..3c30847e0 --- /dev/null +++ b/tests/buffered.rs @@ -0,0 +1,64 @@ +extern crate futures; +extern crate futures_io; +extern crate futures_mio; +extern crate env_logger; + +use std::net::TcpStream; +use std::thread; +use std::io::{Read, Write}; + +use futures::Future; +use futures::stream::Stream; +use futures_io::{BufReader, BufWriter, copy}; + +macro_rules! t { + ($e:expr) => (match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {:?}", stringify!($e), e), + }) +} + +#[test] +fn echo_server() { + const N: usize = 1024; + drop(env_logger::init()); + + let mut l = t!(futures_mio::Loop::new()); + let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); + let srv = t!(l.run(srv)); + let addr = t!(srv.local_addr()); + + let msg = "foo bar baz"; + let t = thread::spawn(move || { + let mut s = t!(TcpStream::connect(&addr)); + + let t2 = thread::spawn(move || { + let mut s = t!(TcpStream::connect(&addr)); + let mut b = vec![0; msg.len() * N]; + t!(s.read_exact(&mut b)); + b + }); + + let mut expected = Vec::::new(); + for _i in 0..N { + expected.extend(msg.as_bytes()); + assert_eq!(t!(s.write(msg.as_bytes())), msg.len()); + } + (expected, t2) + }); + + let clients = srv.incoming().take(2).map(|e| e.0).collect(); + let copied = clients.and_then(|clients| { + let mut clients = clients.into_iter(); + let a = BufReader::new(clients.next().unwrap()); + let b = BufWriter::new(clients.next().unwrap()); + copy(a, b) + }); + + let amt = t!(l.run(copied)); + let (expected, t2) = t.join().unwrap(); + let actual = t2.join().unwrap(); + + assert!(expected == actual); + assert_eq!(amt, msg.len() as u64 * 1024); +} diff --git a/tests/chain.rs b/tests/chain.rs new file mode 100644 index 000000000..fb2c9b8fa --- /dev/null +++ b/tests/chain.rs @@ -0,0 +1,52 @@ +extern crate futures; +extern crate futures_io; +extern crate futures_mio; + +use std::net::TcpStream; +use std::thread; +use std::io::Write; + +use futures::Future; +use futures::stream::Stream; +use futures_io::{chain, read_to_end}; + +macro_rules! t { + ($e:expr) => (match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {:?}", stringify!($e), e), + }) +} + +#[test] +fn chain_clients() { + let mut l = t!(futures_mio::Loop::new()); + let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); + let srv = t!(l.run(srv)); + let addr = t!(srv.local_addr()); + + let t = thread::spawn(move || { + let mut s1 = TcpStream::connect(&addr).unwrap(); + s1.write_all(b"foo ").unwrap(); + let mut s2 = TcpStream::connect(&addr).unwrap(); + s2.write_all(b"bar ").unwrap(); + let mut s3 = TcpStream::connect(&addr).unwrap(); + s3.write_all(b"baz").unwrap(); + }); + + let clients = srv.incoming().map(|e| e.0).take(3); + let copied = clients.collect().and_then(|clients| { + let mut clients = clients.into_iter(); + let a = clients.next().unwrap(); + let b = clients.next().unwrap(); + let c = clients.next().unwrap(); + + let d = chain(a, b); + let d = chain(d, c); + read_to_end(d, Vec::new()) + }); + + let data = t!(l.run(copied)); + t.join().unwrap(); + + assert_eq!(data, b"foo bar baz"); +} diff --git a/tests/echo.rs b/tests/echo.rs new file mode 100644 index 000000000..a9ee4e56c --- /dev/null +++ b/tests/echo.rs @@ -0,0 +1,48 @@ +extern crate futures; +extern crate futures_io; +extern crate futures_mio; + +use std::net::TcpStream; +use std::thread; +use std::io::{Read, Write}; + +use futures::Future; +use futures::stream::Stream; +use futures_io::{copy, TaskIo}; + +macro_rules! t { + ($e:expr) => (match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {:?}", stringify!($e), e), + }) +} + +#[test] +fn echo_server() { + let mut l = t!(futures_mio::Loop::new()); + let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); + let srv = t!(l.run(srv)); + let addr = t!(srv.local_addr()); + + let msg = "foo bar baz"; + let t = thread::spawn(move || { + let mut s = TcpStream::connect(&addr).unwrap(); + + for _i in 0..1024 { + assert_eq!(t!(s.write(msg.as_bytes())), msg.len()); + let mut buf = [0; 1024]; + assert_eq!(t!(s.read(&mut buf)), msg.len()); + assert_eq!(&buf[..msg.len()], msg.as_bytes()); + } + }); + + let clients = srv.incoming(); + let client = clients.into_future().map(|e| e.0.unwrap()).map_err(|e| e.0); + let halves = client.and_then(|s| TaskIo::new(s.0)).map(|i| i.split()); + let copied = halves.and_then(|(a, b)| copy(a, b)); + + 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 new file mode 100644 index 000000000..d7082f459 --- /dev/null +++ b/tests/limit.rs @@ -0,0 +1,44 @@ +extern crate futures; +extern crate futures_io; +extern crate futures_mio; + +use std::net::TcpStream; +use std::thread; +use std::io::Write; + +use futures::Future; +use futures::stream::Stream; +use futures_io::{read_to_end, take}; + +macro_rules! t { + ($e:expr) => (match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {:?}", stringify!($e), e), + }) +} + +#[test] +fn limit() { + let mut l = t!(futures_mio::Loop::new()); + let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); + let srv = t!(l.run(srv)); + let addr = t!(srv.local_addr()); + + let t = thread::spawn(move || { + let mut s1 = TcpStream::connect(&addr).unwrap(); + s1.write_all(b"foo bar baz").unwrap(); + }); + + let clients = srv.incoming().map(|e| e.0).take(1); + let copied = clients.collect().and_then(|clients| { + let mut clients = clients.into_iter(); + let a = clients.next().unwrap(); + + read_to_end(take(a, 4), Vec::new()) + }); + + let data = t!(l.run(copied)); + t.join().unwrap(); + + assert_eq!(data, b"foo "); +} diff --git a/tests/tcp.rs b/tests/tcp.rs new file mode 100644 index 000000000..a984977ab --- /dev/null +++ b/tests/tcp.rs @@ -0,0 +1,81 @@ +extern crate futures; +extern crate futures_mio; + +use std::net::{TcpListener, TcpStream}; +use std::sync::mpsc::channel; +use std::thread; + +use futures::Future; +use futures::stream::Stream; + +macro_rules! t { + ($e:expr) => (match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {:?}", stringify!($e), e), + }) +} + +#[test] +fn connect() { + let mut l = t!(futures_mio::Loop::new()); + let srv = t!(TcpListener::bind("127.0.0.1:0")); + let addr = t!(srv.local_addr()); + let t = thread::spawn(move || { + t!(srv.accept()).0 + }); + + let stream = l.handle().tcp_connect(&addr); + let mine = t!(l.run(stream)); + let theirs = t.join().unwrap(); + + assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); + assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); +} + +#[test] +fn accept() { + let mut l = t!(futures_mio::Loop::new()); + let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); + let srv = t!(l.run(srv)); + let addr = t!(srv.local_addr()); + + let (tx, rx) = channel(); + let client = srv.incoming().map(move |t| { + tx.send(()).unwrap(); + t.0 + }).into_future().map_err(|e| e.0); + assert!(rx.try_recv().is_err()); + let t = thread::spawn(move || { + TcpStream::connect(&addr).unwrap() + }); + + let (mine, _remaining) = t!(l.run(client)); + let mine = mine.unwrap(); + let theirs = t.join().unwrap(); + + assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr())); + assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr())); +} + +#[test] +fn accept2() { + let mut l = t!(futures_mio::Loop::new()); + let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); + let srv = t!(l.run(srv)); + let addr = t!(srv.local_addr()); + + let t = thread::spawn(move || { + TcpStream::connect(&addr).unwrap() + }); + + let (tx, rx) = channel(); + let client = srv.incoming().map(move |t| { + tx.send(()).unwrap(); + t.0 + }).into_future().map_err(|e| e.0); + assert!(rx.try_recv().is_err()); + + let (mine, _remaining) = t!(l.run(client)); + mine.unwrap(); + t.join().unwrap(); +}