Let's rename everything!

Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "futures-mio"
+version = "0.1.0"
+authors = ["Alex Crichton <[email protected]>"]
+
+[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

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::<SocketAddr>().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();
+}

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::<SocketAddr>().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<futures_mio::IoFuture<()>> {
+    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()
+}

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<Vec<u8>>,
+    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<R, F>(&mut self, f: F) -> R
+        where F: FnOnce(&mut Vec<u8>) -> 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<R: Read>(&mut self, socket: &mut R) -> io::Result<(usize, bool)> {
+        unsafe fn slice_to_end(v: &mut Vec<u8>) -> &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<R> {
+    source: R,
+    source_ready: ReadinessStream,
+    read_ready: bool,
+    buf: InputBuf,
+}
+
+impl<R: Read + Send + 'static> BufReader<R> {
+    pub fn new(source: R, source_ready: ReadinessStream) -> BufReader<R> {
+        BufReader {
+            source: source,
+            source_ready: source_ready,
+            read_ready: false,
+            buf: InputBuf::new(),
+        }
+    }
+
+    pub fn buf(&mut self) -> &mut InputBuf {
+        &mut self.buf
+    }
+}
+
+impl<R: Read + Send + 'static> Stream for BufReader<R> {
+    type Item = ();
+    type Error = io::Error;
+
+    fn poll(&mut self, task: &mut Task) -> Poll<Option<()>, 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)
+    }
+}

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<W> {
+    sink: W,
+    sink_ready: ReadinessStream,
+    write_ready: bool,
+    buf: Vec<u8>,
+}
+
+impl<W: Write + Send + 'static> BufWriter<W> {
+    pub fn new(sink: W, sink_ready: ReadinessStream) -> BufWriter<W> {
+        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<W> {
+        Flush { writer: Some(self) }
+    }
+
+    pub fn reserve(self, amt: usize) -> Reserve<W> {
+        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<W: Write> Write for BufWriter<W> {
+    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
+        extend(&mut self.buf, data);
+        Ok(data.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        // TODO: something reasonable
+        unimplemented!()
+    }
+}
+
+pub struct Flush<W> {
+    writer: Option<BufWriter<W>>,
+}
+
+impl<W: Write + Send + 'static> Flush<W> {
+    pub fn is_dirty(&self) -> bool {
+        self.writer.as_ref().unwrap().is_dirty()
+    }
+
+    pub fn into_inner(mut self) -> BufWriter<W> {
+        self.writer.take().unwrap()
+    }
+}
+
+impl<W: Write + Send + 'static> Future for Flush<W> {
+    type Item = BufWriter<W>;
+    type Error = (io::Error, BufWriter<W>);
+
+    fn poll(&mut self, task: &mut Task)
+            -> Poll<BufWriter<W>, (io::Error, BufWriter<W>)> {
+        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<u8>, 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<W> {
+    amt: usize,
+    writer: Option<BufWriter<W>>,
+}
+
+impl<W: Write + Send + 'static> Future for Reserve<W> {
+    type Item = BufWriter<W>;
+    type Error = (io::Error, BufWriter<W>);
+
+    fn poll(&mut self, task: &mut Task)
+            -> Poll<BufWriter<W>, (io::Error, BufWriter<W>)> {
+        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)
+    }
+}

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<bool>,
+    io: mio::Poll,
+    tx: mio::channel::Sender<Message>,
+    rx: mio::channel::Receiver<Message>,
+    dispatch: RefCell<Slab<Scheduled, usize>>,
+}
+
+/// 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<Message>,
+}
+
+struct Scheduled {
+    source: IoSource,
+    waiter: Option<TaskHandle>,
+}
+
+enum Message {
+    AddSource(IoSource, Arc<Slot<io::Result<usize>>>),
+    DropSource(usize),
+    Schedule(usize, TaskHandle),
+    Deschedule(usize),
+    Shutdown,
+}
+
+pub struct Source<E: ?Sized> {
+    readiness: AtomicUsize,
+    io: E,
+}
+
+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.
+    pub fn new() -> io::Result<Loop> {
+        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<F: Future>(&mut self, f: F) -> Result<F::Item, F::Error> {
+        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<usize> {
+        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<F, R>(&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<Slot<io::Result<usize>>>) {
+        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<IoSource>,
+    result: Option<(Arc<Slot<io::Result<usize>>>, slot::Token)>,
+}
+
+impl Future for AddSource {
+    type Item = usize;
+    type Error = io::Error;
+
+    fn poll(&mut self, _task: &mut Task) -> Poll<usize, io::Error> {
+        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<E> Source<E> {
+    pub fn new(e: E) -> Source<E> {
+        Source {
+            readiness: AtomicUsize::new(0),
+            io: e,
+        }
+    }
+}
+
+impl<E: ?Sized> Source<E> {
+    pub fn take_readiness(&self) -> Option<Ready> {
+        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
+    }
+}

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<T> = Future<Item=T, Error=io::Error>;
+
+/// A convenience typedef around a `Stream` whose error component is `io::Error`
+pub type IoStream<T> = Stream<Item=T, Error=io::Error>;
+
+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};

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<DropSource>,
+}
+
+impl ReadinessStream {
+    pub fn new(loop_handle: LoopHandle, source: IoSource)
+               -> Box<IoFuture<ReadinessStream>> {
+        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<Option<Ready>, 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)
+    }
+}

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<Source<mio::tcp::TcpListener>>,
+}
+
+impl TcpListener {
+    fn new(listener: mio::tcp::TcpListener,
+           handle: LoopHandle) -> Box<IoFuture<TcpListener>> {
+        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<IoFuture<TcpListener>> {
+        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<SocketAddr> {
+        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<IoStream<(TcpStream, SocketAddr)>> {
+        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<Source<mio::tcp::TcpListener>>,
+}
+
+impl Iterator for NonblockingIter {
+    type Item = io::Result<(mio::tcp::TcpStream, SocketAddr)>;
+
+    fn next(&mut self) -> Option<io::Result<(mio::tcp::TcpStream, SocketAddr)>> {
+        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<Option<Ready>, 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<Source<mio::tcp::TcpStream>>,
+    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<IoFuture<TcpListener>> {
+        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<IoFuture<TcpStream>> {
+        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<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,
+            })
+        }).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<IoFuture<TcpStream>> {
+        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<SocketAddr> {
+        self.source.io().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()
+    }
+}
+
+impl Future for TcpStreamNew {
+    type Item = TcpStream;
+    type Error = io::Error;
+
+    fn poll(&mut self, task: &mut Task) -> Poll<TcpStream, io::Error> {
+        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<usize> {
+        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<usize> {
+        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<usize> {
+        self.source.io().read(buf)
+    }
+}
+
+impl<'a> Write for &'a TcpStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        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<Option<Ready>, io::Error> {
+        self.ready.poll(task)
+    }
+
+    fn schedule(&mut self, task: &mut Task) {
+        self.ready.schedule(task)
+    }
+}

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::<u8>::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);
+}

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");
+}

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);
+}

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 ");
+}

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();
+}