Initial commit

.gitignore

@@ -0,0 +1,2 @@
+/target
+Cargo.lock

Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "async-io"
+version = "0.1.0"
+authors = ["Stjepan Glavina <stjepang@gmail.com>"]
+edition = "2018"
+
+[dependencies]
+blocking = "0.4.6"
+concurrent-queue = "1.1.1"
+futures-io = { version = "0.3.5", default-features = false, features = ["std"] }
+futures-util = { version = "0.3.5", default-features = false, features = ["std", "io"] }
+libc = "0.2.71"
+once_cell = "1.4.0"
+parking = "1.0.3"
+slab = "0.4.2"
+socket2 = { version = "0.3.12", features = ["pair", "unix"] }
+
+[target.'cfg(windows)'.dependencies]
+wepoll-sys-stjepang = "1.0.6"
+winapi = { version = "0.3.8", features = ["ioapiset"] }
+
+[dev-dependencies]
+async-channel = "1.1.1"
+async-dup = "1.1.0"
+futures = { version = "0.3.5", default-features = false, features = ["std"] }
+tempfile = "3.1.0"

src/sys.rs

@@ -0,0 +1,277 @@
+#[cfg(unix)]
+fn check_err(res: libc::c_int) -> Result<libc::c_int, std::io::Error> {
+    if res == -1 {
+        return Err(std::io::Error::last_os_error());
+    }
+
+    Ok(res)
+}
+
+#[cfg(any(
+    target_os = "macos",
+    target_os = "ios",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "dragonfly",
+))]
+/// Kqueue.
+pub mod event {
+    use super::check_err;
+    use std::os::unix::io::RawFd;
+
+    #[cfg(any(
+        target_os = "macos",
+        target_os = "ios",
+        target_os = "freebsd",
+        target_os = "dragonfly",
+        target_os = "openbsd"
+    ))]
+    #[allow(non_camel_case_types)]
+    type type_of_nchanges = libc::c_int;
+    #[cfg(target_os = "netbsd")]
+    #[allow(non_camel_case_types)]
+    type type_of_nchanges = libc::size_t;
+
+    #[cfg(target_os = "netbsd")]
+    #[allow(non_camel_case_types)]
+    type type_of_event_filter = u32;
+    #[cfg(not(target_os = "netbsd"))]
+    #[allow(non_camel_case_types)]
+    type type_of_event_filter = i16;
+
+    #[cfg(any(
+        target_os = "dragonfly",
+        target_os = "freebsd",
+        target_os = "ios",
+        target_os = "macos",
+        target_os = "openbsd"
+    ))]
+    #[allow(non_camel_case_types)]
+    type type_of_udata = *mut libc::c_void;
+    #[cfg(any(
+        target_os = "dragonfly",
+        target_os = "freebsd",
+        target_os = "ios",
+        target_os = "macos"
+    ))]
+    #[allow(non_camel_case_types)]
+    type type_of_data = libc::intptr_t;
+    #[cfg(any(target_os = "netbsd"))]
+    #[allow(non_camel_case_types)]
+    type type_of_udata = libc::intptr_t;
+    #[cfg(any(target_os = "netbsd", target_os = "openbsd"))]
+    #[allow(non_camel_case_types)]
+    type type_of_data = libc::int64_t;
+
+    #[derive(Clone, Copy)]
+    #[repr(C)]
+    pub struct KEvent(libc::kevent);
+
+    unsafe impl Send for KEvent {}
+
+    impl KEvent {
+        pub fn new(
+            ident: libc::uintptr_t,
+            filter: EventFilter,
+            flags: EventFlag,
+            fflags: FilterFlag,
+            data: libc::intptr_t,
+            udata: libc::intptr_t,
+        ) -> KEvent {
+            KEvent(libc::kevent {
+                ident,
+                filter: filter as type_of_event_filter,
+                flags,
+                fflags,
+                data: data as type_of_data,
+                udata: udata as type_of_udata,
+            })
+        }
+
+        pub fn filter(&self) -> EventFilter {
+            unsafe { std::mem::transmute(self.0.filter as type_of_event_filter) }
+        }
+
+        pub fn flags(&self) -> EventFlag {
+            self.0.flags
+        }
+
+        pub fn data(&self) -> libc::intptr_t {
+            self.0.data as libc::intptr_t
+        }
+
+        pub fn udata(&self) -> libc::intptr_t {
+            self.0.udata as libc::intptr_t
+        }
+    }
+
+    #[cfg(any(
+        target_os = "dragonfly",
+        target_os = "freebsd",
+        target_os = "ios",
+        target_os = "macos",
+        target_os = "openbsd"
+    ))]
+    pub type EventFlag = u16;
+    #[cfg(any(target_os = "netbsd"))]
+    pub type EventFlag = u32;
+
+    pub type FilterFlag = u32;
+
+    #[cfg(target_os = "netbsd")]
+    pub type EventFilter = u32;
+    #[cfg(not(target_os = "netbsd"))]
+    pub type EventFilter = i16;
+
+    pub fn kqueue() -> Result<RawFd, std::io::Error> {
+        let res = unsafe { libc::kqueue() };
+
+        check_err(res)
+    }
+
+    pub fn kevent_ts(
+        kq: RawFd,
+        changelist: &[KEvent],
+        eventlist: &mut [KEvent],
+        timeout_opt: Option<libc::timespec>,
+    ) -> Result<usize, std::io::Error> {
+        let res = unsafe {
+            libc::kevent(
+                kq,
+                changelist.as_ptr() as *const libc::kevent,
+                changelist.len() as type_of_nchanges,
+                eventlist.as_mut_ptr() as *mut libc::kevent,
+                eventlist.len() as type_of_nchanges,
+                if let Some(ref timeout) = timeout_opt {
+                    timeout as *const libc::timespec
+                } else {
+                    std::ptr::null()
+                },
+            )
+        };
+
+        check_err(res).map(|r| r as usize)
+    }
+}
+
+#[cfg(any(target_os = "linux", target_os = "android", target_os = "illumos"))]
+/// Epoll.
+pub mod epoll {
+    use super::check_err;
+    use std::os::unix::io::RawFd;
+
+    #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+    #[repr(i32)]
+    pub enum EpollOp {
+        EpollCtlAdd = libc::EPOLL_CTL_ADD,
+        EpollCtlDel = libc::EPOLL_CTL_DEL,
+        EpollCtlMod = libc::EPOLL_CTL_MOD,
+    }
+
+    pub type EpollFlags = libc::c_int;
+
+    pub fn epoll_create1() -> Result<RawFd, std::io::Error> {
+        // According to libuv, `EPOLL_CLOEXEC` is not defined on Android API < 21.
+        // But `EPOLL_CLOEXEC` is an alias for `O_CLOEXEC` on that platform, so we use it instead.
+        #[cfg(target_os = "android")]
+        const CLOEXEC: libc::c_int = libc::O_CLOEXEC;
+        #[cfg(not(target_os = "android"))]
+        const CLOEXEC: libc::c_int = libc::EPOLL_CLOEXEC;
+
+        let fd = unsafe {
+            // Check if the `epoll_create1` symbol is available on this platform.
+            let ptr = libc::dlsym(
+                libc::RTLD_DEFAULT,
+                "epoll_create1\0".as_ptr() as *const libc::c_char,
+            );
+
+            if ptr.is_null() {
+                // If not, use `epoll_create` and manually set `CLOEXEC`.
+                let fd = check_err(libc::epoll_create(1024))?;
+                let flags = libc::fcntl(fd, libc::F_GETFD);
+                libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC);
+                fd
+            } else {
+                // Use `epoll_create1` with `CLOEXEC`.
+                let epoll_create1 = std::mem::transmute::<
+                    *mut libc::c_void,
+                    unsafe extern "C" fn(libc::c_int) -> libc::c_int,
+                >(ptr);
+                check_err(epoll_create1(CLOEXEC))?
+            }
+        };
+
+        Ok(fd)
+    }
+
+    pub fn epoll_ctl<'a, T>(
+        epfd: RawFd,
+        op: EpollOp,
+        fd: RawFd,
+        event: T,
+    ) -> Result<(), std::io::Error>
+    where
+        T: Into<Option<&'a mut EpollEvent>>,
+    {
+        let mut event: Option<&mut EpollEvent> = event.into();
+        if event.is_none() && op != EpollOp::EpollCtlDel {
+            Err(std::io::Error::from_raw_os_error(libc::EINVAL))
+        } else {
+            let res = unsafe {
+                if let Some(ref mut event) = event {
+                    libc::epoll_ctl(epfd, op as libc::c_int, fd, &mut event.event)
+                } else {
+                    libc::epoll_ctl(epfd, op as libc::c_int, fd, std::ptr::null_mut())
+                }
+            };
+            check_err(res).map(drop)
+        }
+    }
+
+    pub fn epoll_wait(
+        epfd: RawFd,
+        events: &mut [EpollEvent],
+        timeout_ms: isize,
+    ) -> Result<usize, std::io::Error> {
+        let res = unsafe {
+            libc::epoll_wait(
+                epfd,
+                events.as_mut_ptr() as *mut libc::epoll_event,
+                events.len() as libc::c_int,
+                timeout_ms as libc::c_int,
+            )
+        };
+
+        check_err(res).map(|r| r as usize)
+    }
+
+    #[derive(Clone, Copy)]
+    #[repr(transparent)]
+    pub struct EpollEvent {
+        event: libc::epoll_event,
+    }
+
+    impl EpollEvent {
+        pub fn new(events: EpollFlags, data: u64) -> Self {
+            EpollEvent {
+                event: libc::epoll_event {
+                    events: events as u32,
+                    u64: data,
+                },
+            }
+        }
+
+        pub fn empty() -> Self {
+            unsafe { std::mem::zeroed::<EpollEvent>() }
+        }
+
+        pub fn events(&self) -> EpollFlags {
+            self.event.events as libc::c_int
+        }
+
+        pub fn data(&self) -> u64 {
+            self.event.u64
+        }
+    }
+}

tests/async.rs

@@ -0,0 +1,339 @@
+use std::future::Future;
+use std::io;
+use std::net::{Shutdown, TcpListener, TcpStream, UdpSocket};
+#[cfg(unix)]
+use std::os::unix::net::{UnixDatagram, UnixListener, UnixStream};
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+use async_io::{Async, Timer};
+use blocking::block_on;
+use futures::{AsyncReadExt, AsyncWriteExt, StreamExt};
+#[cfg(unix)]
+use tempfile::tempdir;
+
+const LOREM_IPSUM: &[u8] = b"
+Lorem ipsum dolor sit amet, consectetur adipiscing elit.
+Donec pretium ante erat, vitae sodales mi varius quis.
+Etiam vestibulum lorem vel urna tempor, eu fermentum odio aliquam.
+Aliquam consequat urna vitae ipsum pulvinar, in blandit purus eleifend.
+";
+
+fn spawn<T: Send + 'static>(
+    f: impl Future<Output = T> + Send + 'static,
+) -> impl Future<Output = T> + Send + 'static {
+    let (s, r) = async_channel::bounded(1);
+
+    thread::spawn(move || {
+        block_on(async {
+            let _ = s.send(f.await).await;
+        })
+    });
+
+    Box::pin(async move { r.recv().await.unwrap() })
+}
+
+#[test]
+fn tcp_connect() -> io::Result<()> {
+    block_on(async {
+        let listener = Async::<TcpListener>::bind("127.0.0.1:0")?;
+        let addr = listener.get_ref().local_addr()?;
+        let task = spawn(async move { listener.accept().await });
+
+        let stream2 = Async::<TcpStream>::connect(&addr).await?;
+        let stream1 = task.await?.0;
+
+        assert_eq!(
+            stream1.get_ref().peer_addr()?,
+            stream2.get_ref().local_addr()?,
+        );
+        assert_eq!(
+            stream2.get_ref().peer_addr()?,
+            stream1.get_ref().local_addr()?,
+        );
+
+        // Now that the listener is closed, connect should fail.
+        let err = Async::<TcpStream>::connect(&addr).await.unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::ConnectionRefused);
+
+        Ok(())
+    })
+}
+
+#[test]
+fn tcp_peek_read() -> io::Result<()> {
+    block_on(async {
+        let listener = Async::<TcpListener>::bind("127.0.0.1:0")?;
+        let addr = listener.get_ref().local_addr()?;
+
+        let mut stream = Async::<TcpStream>::connect(addr).await?;
+        stream.write_all(LOREM_IPSUM).await?;
+
+        let mut buf = [0; 1024];
+        let mut incoming = listener.incoming();
+        let mut stream = incoming.next().await.unwrap()?;
+
+        let n = stream.peek(&mut buf).await?;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+        let n = stream.read(&mut buf).await?;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+
+        Ok(())
+    })
+}
+
+#[test]
+fn tcp_reader_hangup() -> io::Result<()> {
+    block_on(async {
+        let listener = Async::<TcpListener>::bind("127.0.0.1:0")?;
+        let addr = listener.get_ref().local_addr()?;
+        let task = spawn(async move { listener.accept().await });
+
+        let mut stream2 = Async::<TcpStream>::connect(&addr).await?;
+        let stream1 = task.await?.0;
+
+        let task = spawn(async move {
+            Timer::after(Duration::from_secs(1)).await;
+            drop(stream1);
+        });
+
+        while stream2.write_all(LOREM_IPSUM).await.is_ok() {}
+        task.await;
+
+        Ok(())
+    })
+}
+
+#[test]
+fn tcp_writer_hangup() -> io::Result<()> {
+    block_on(async {
+        let listener = Async::<TcpListener>::bind("127.0.0.1:0")?;
+        let addr = listener.get_ref().local_addr()?;
+        let task = spawn(async move { listener.accept().await });
+
+        let mut stream2 = Async::<TcpStream>::connect(&addr).await?;
+        let stream1 = task.await?.0;
+
+        let task = spawn(async move {
+            Timer::after(Duration::from_secs(1)).await;
+            drop(stream1);
+        });
+
+        let mut v = vec![];
+        stream2.read_to_end(&mut v).await?;
+        assert!(v.is_empty());
+
+        task.await;
+        Ok(())
+    })
+}
+
+#[test]
+fn udp_send_recv() -> io::Result<()> {
+    block_on(async {
+        let socket1 = Async::<UdpSocket>::bind("127.0.0.1:0")?;
+        let socket2 = Async::<UdpSocket>::bind("127.0.0.1:0")?;
+        socket1.get_ref().connect(socket2.get_ref().local_addr()?)?;
+
+        let mut buf = [0u8; 1024];
+
+        socket1.send(LOREM_IPSUM).await?;
+        let n = socket2.peek(&mut buf).await?;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+        let n = socket2.recv(&mut buf).await?;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+
+        socket2
+            .send_to(LOREM_IPSUM, socket1.get_ref().local_addr()?)
+            .await?;
+        let n = socket1.peek_from(&mut buf).await?.0;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+        let n = socket1.recv_from(&mut buf).await?.0;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+
+        Ok(())
+    })
+}
+
+#[cfg(unix)]
+#[test]
+fn udp_connect() -> io::Result<()> {
+    block_on(async {
+        let dir = tempdir()?;
+        let path = dir.path().join("socket");
+
+        let listener = Async::<UnixListener>::bind(&path)?;
+
+        let mut stream = Async::<UnixStream>::connect(&path).await?;
+        stream.write_all(LOREM_IPSUM).await?;
+
+        let mut buf = [0; 1024];
+        let mut incoming = listener.incoming();
+        let mut stream = incoming.next().await.unwrap()?;
+
+        let n = stream.read(&mut buf).await?;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+
+        Ok(())
+    })
+}
+
+#[cfg(unix)]
+#[test]
+fn uds_connect() -> io::Result<()> {
+    block_on(async {
+        let dir = tempdir()?;
+        let path = dir.path().join("socket");
+        let listener = Async::<UnixListener>::bind(&path)?;
+
+        let addr = listener.get_ref().local_addr()?;
+        let task = spawn(async move { listener.accept().await });
+
+        let stream2 = Async::<UnixStream>::connect(addr.as_pathname().unwrap()).await?;
+        let stream1 = task.await?.0;
+
+        assert_eq!(
+            stream1.get_ref().peer_addr()?.as_pathname(),
+            stream2.get_ref().local_addr()?.as_pathname(),
+        );
+        assert_eq!(
+            stream2.get_ref().peer_addr()?.as_pathname(),
+            stream1.get_ref().local_addr()?.as_pathname(),
+        );
+
+        // Now that the listener is closed, connect should fail.
+        let err = Async::<UnixStream>::connect(addr.as_pathname().unwrap())
+            .await
+            .unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::ConnectionRefused);
+
+        Ok(())
+    })
+}
+
+#[cfg(unix)]
+#[test]
+fn uds_send_recv() -> io::Result<()> {
+    block_on(async {
+        let (socket1, socket2) = Async::<UnixDatagram>::pair()?;
+
+        socket1.send(LOREM_IPSUM).await?;
+        let mut buf = [0; 1024];
+        let n = socket2.recv(&mut buf).await?;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+
+        Ok(())
+    })
+}
+
+#[cfg(unix)]
+#[test]
+fn uds_send_to_recv_from() -> io::Result<()> {
+    block_on(async {
+        let dir = tempdir()?;
+        let path = dir.path().join("socket");
+        let socket1 = Async::<UnixDatagram>::bind(&path)?;
+        let socket2 = Async::<UnixDatagram>::unbound()?;
+
+        socket2.send_to(LOREM_IPSUM, &path).await?;
+        let mut buf = [0; 1024];
+        let n = socket1.recv_from(&mut buf).await?.0;
+        assert_eq!(&buf[..n], LOREM_IPSUM);
+
+        Ok(())
+    })
+}
+
+#[cfg(unix)]
+#[test]
+fn uds_reader_hangup() -> io::Result<()> {
+    block_on(async {
+        let (socket1, mut socket2) = Async::<UnixStream>::pair()?;
+
+        let task = spawn(async move {
+            Timer::after(Duration::from_secs(1)).await;
+            drop(socket1);
+        });
+
+        while socket2.write_all(LOREM_IPSUM).await.is_ok() {}
+        task.await;
+
+        Ok(())
+    })
+}
+
+#[cfg(unix)]
+#[test]
+fn uds_writer_hangup() -> io::Result<()> {
+    block_on(async {
+        let (socket1, mut socket2) = Async::<UnixStream>::pair()?;
+
+        let task = spawn(async move {
+            Timer::after(Duration::from_secs(1)).await;
+            drop(socket1);
+        });
+
+        let mut v = vec![];
+        socket2.read_to_end(&mut v).await?;
+        assert!(v.is_empty());
+
+        task.await;
+        Ok(())
+    })
+}
+
+// Test that we correctly re-register interests when we are previously
+// interested in both readable and writable events and then we get only one of
+// them. (we need to re-register interest on the other.)
+#[test]
+fn tcp_duplex() -> io::Result<()> {
+    block_on(async {
+        let listener = Async::<TcpListener>::bind("127.0.0.1:0")?;
+        let stream0 =
+            Arc::new(Async::<TcpStream>::connect(listener.get_ref().local_addr()?).await?);
+        let stream1 = Arc::new(listener.accept().await?.0);
+
+        async fn do_read(s: Arc<Async<TcpStream>>) -> io::Result<()> {
+            let mut buf = vec![0u8; 4096];
+            loop {
+                let len = (&*s).read(&mut buf).await?;
+                if len == 0 {
+                    return Ok(());
+                }
+            }
+        }
+
+        async fn do_write(s: Arc<Async<TcpStream>>) -> io::Result<()> {
+            let buf = vec![0u8; 4096];
+            for _ in 0..4096 {
+                (&*s).write_all(&buf).await?;
+            }
+            s.get_ref().shutdown(Shutdown::Write)?;
+            Ok(())
+        }
+
+        // Read from and write to stream0.
+        let r0 = spawn(do_read(stream0.clone()));
+        let w0 = spawn(do_write(stream0));
+
+        // Sleep a bit, so that reading and writing are both blocked.
+        Timer::after(Duration::from_millis(5)).await;
+
+        // Start reading stream1, make stream0 writable.
+        let r1 = spawn(do_read(stream1.clone()));
+
+        // Finish writing to stream0.
+        w0.await?;
+        r1.await?;
+
+        // Start writing to stream1, make stream0 readable.
+        let w1 = spawn(do_write(stream1));
+
+        // Will r0 be correctly woken?
+        r0.await?;
+        w1.await?;
+
+        Ok(())
+    })
+}

tests/timer.rs

@@ -0,0 +1,27 @@
+use std::time::{Duration, Instant};
+
+use async_io::Timer;
+use blocking::block_on;
+
+#[test]
+fn timer_at() {
+    let before = block_on(async {
+        let now = Instant::now();
+        let when = now + Duration::from_secs(1);
+        Timer::at(when).await;
+        now
+    });
+
+    assert!(before.elapsed() >= Duration::from_secs(1));
+}
+
+#[test]
+fn timer_after() {
+    let before = block_on(async {
+        let now = Instant::now();
+        Timer::after(Duration::from_secs(1)).await;
+        now
+    });
+
+    assert!(before.elapsed() >= Duration::from_secs(1));
+}