Merge #55

55: chore: bump tokio to v1 r=ryankurte a=oll3

Bump tokio to version 1.x.
Use tokio AsyncFd since tokio no longer exposes mio internals.

Co-authored-by: Olle Sandberg <[email protected]>

.travis.yml

@@ -72,7 +72,7 @@ matrix:
 
     # MSRV
     - env: TARGET=x86_64-unknown-linux-gnu DISABLE_TESTS=1
-      rust: 1.39.0
+      rust: 1.45.0
       if: (branch = staging OR branch = trying) OR (type = pull_request AND branch = master)
 
 before_install:

CHANGELOG.md

@@ -8,6 +8,13 @@ Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+## v0.5.0
+
+- Update Tokio to 1.x. #[55]((https://github.com/rust-embedded/gpio-cdev/pull/55).
+- Breaking change of `LineEventHandle::get_event()` which now expects `&mut self`.
+- MSRV is now 1.45.0
+
+
 ## v0.4.0 - 2020-08-01
 
 - Removed pub "errors" module.  Error now exposed at top level.

Cargo.toml

@@ -13,7 +13,7 @@ edition = "2018"
 
 [features]
 default = []
-async-tokio = ["tokio", "futures", "mio"]
+async-tokio = ["tokio", "futures"]
 
 [[example]]
 name = "async_tokio"
@@ -23,13 +23,13 @@ required-features = ["async-tokio"]
 bitflags = "1.0"
 libc = "0.2"
 nix = "0.14"
-tokio = { version = "0.2", features = ["io-driver", "rt-threaded", "macros"], optional = true }
+tokio = { version = "1", features = ["io-std", "net"], optional = true }
 futures = { version = "0.3", optional = true }
-mio = { version = "0.6", optional = true }
 
 [dev-dependencies]
 quicli = "0.2"
 anyhow = "1.0"
+tokio = { version = "1", features = ["io-std", "rt-multi-thread", "macros", "net"] }
 
 [package.metadata.docs.rs]
 # To build locally:

README.md

@@ -206,7 +206,7 @@ to be considered reliable.
 
 ## Minimum Supported Rust Version (MSRV)
 
-This crate is guaranteed to compile on stable Rust 1.39.0 and up. It *might*
+This crate is guaranteed to compile on stable Rust 1.45.0 and up. It *might*
 compile with older versions but that may change in any new patch release.
 
 ## License

src/async_tokio.rs

@@ -11,48 +11,14 @@
 use futures::ready;
 use futures::stream::Stream;
 use futures::task::{Context, Poll};
-use mio::event::Evented;
-use mio::unix::EventedFd;
-use mio::{PollOpt, Ready, Token};
-use tokio::io::PollEvented;
+use tokio::io::unix::{AsyncFd, TryIoError};
 
-use std::io;
 use std::os::unix::io::AsRawFd;
 use std::pin::Pin;
 
 use super::event_err;
 use super::{LineEvent, LineEventHandle, Result};
 
-struct PollWrapper {
-    handle: LineEventHandle,
-}
-
-impl Evented for PollWrapper {
-    fn register(
-        &self,
-        poll: &mio::Poll,
-        token: Token,
-        interest: Ready,
-        opts: PollOpt,
-    ) -> io::Result<()> {
-        EventedFd(&self.handle.file.as_raw_fd()).register(poll, token, interest, opts)
-    }
-
-    fn reregister(
-        &self,
-        poll: &mio::Poll,
-        token: Token,
-        interest: Ready,
-        opts: PollOpt,
-    ) -> io::Result<()> {
-        EventedFd(&self.handle.file.as_raw_fd()).reregister(poll, token, interest, opts)
-    }
-
-    fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
-        EventedFd(&self.handle.file.as_raw_fd()).deregister(poll)
-    }
-}
-
 /// Wrapper around a `LineEventHandle` which implements a `futures::stream::Stream` for interrupts.
 ///
 /// # Example
@@ -88,7 +54,7 @@ impl Evented for PollWrapper {
 /// # }
 /// ```
 pub struct AsyncLineEventHandle {
-    evented: PollEvented<PollWrapper>,
+    asyncfd: AsyncFd<LineEventHandle>,
 }
 
 impl AsyncLineEventHandle {
@@ -106,36 +72,35 @@ impl AsyncLineEventHandle {
         }
 
         Ok(AsyncLineEventHandle {
-            evented: PollEvented::new(PollWrapper { handle })?,
+            asyncfd: AsyncFd::new(handle)?,
         })
     }
 }
 
 impl Stream for AsyncLineEventHandle {
     type Item = Result<LineEvent>;
 
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
-        let ready = Ready::readable();
-        if let Err(e) = ready!(self.evented.poll_read_ready(cx, ready)) {
-            return Poll::Ready(Some(Err(e.into())));
-        }
-
-        match self.evented.get_ref().handle.read_event() {
-            Ok(Some(event)) => Poll::Ready(Some(Ok(event))),
-            Ok(None) => Poll::Ready(Some(Err(event_err(nix::Error::Sys(
-                nix::errno::Errno::EIO,
-            ))))),
-            Err(nix::Error::Sys(nix::errno::Errno::EAGAIN)) => {
-                self.evented.clear_read_ready(cx, ready)?;
-                Poll::Pending
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        loop {
+            let mut guard = ready!(self.asyncfd.poll_read_ready_mut(cx))?;
+            match guard.try_io(|inner| inner.get_mut().read_event()) {
+                Err(TryIoError { .. }) => {
+                    // Continue
+                }
+                Ok(Ok(Some(event))) => return Poll::Ready(Some(Ok(event))),
+                Ok(Ok(None)) => {
+                    return Poll::Ready(Some(Err(event_err(nix::Error::Sys(
+                        nix::errno::Errno::EIO,
+                    )))))
+                }
+                Ok(Err(err)) => return Poll::Ready(Some(Err(err.into()))),
             }
-            Err(e) => Poll::Ready(Some(Err(event_err(e)))),
         }
     }
 }
 
 impl AsRef<LineEventHandle> for AsyncLineEventHandle {
     fn as_ref(&self) -> &LineEventHandle {
-        &self.evented.get_ref().handle
+        &self.asyncfd.get_ref()
     }
 }

src/lib.rs

@@ -92,6 +92,7 @@ extern crate nix;
 use std::cmp::min;
 use std::ffi::CStr;
 use std::fs::{read_dir, File, ReadDir};
+use std::io::Read;
 use std::mem;
 use std::ops::Index;
 use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
@@ -954,11 +955,11 @@ impl LineEventHandle {
     /// This blocks while there is not another event available from the
     /// kernel for the line which matches the subscription criteria
     /// specified in the `event_flags` when the handle was created.
-    pub fn get_event(&self) -> Result<LineEvent> {
+    pub fn get_event(&mut self) -> Result<LineEvent> {
         match self.read_event() {
             Ok(Some(event)) => Ok(event),
             Ok(None) => Err(event_err(nix::Error::Sys(nix::errno::Errno::EIO))),
-            Err(e) => Err(event_err(e)),
+            Err(e) => Err(e.into()),
         }
     }
 
@@ -981,19 +982,15 @@ impl LineEventHandle {
 
     /// Helper function which returns the line event if a complete event was read, Ok(None) if not
     /// enough data was read or the error returned by `read()`.
-    ///
-    /// This function allows access to the raw `nix::Error` as required, for example, to theck
-    /// whether read() returned -EAGAIN.
-    pub(crate) fn read_event(&self) -> std::result::Result<Option<LineEvent>, nix::Error> {
+    pub(crate) fn read_event(&mut self) -> std::io::Result<Option<LineEvent>> {
         let mut data: ffi::gpioevent_data = unsafe { mem::zeroed() };
         let mut data_as_buf = unsafe {
             slice::from_raw_parts_mut(
                 &mut data as *mut ffi::gpioevent_data as *mut u8,
                 mem::size_of::<ffi::gpioevent_data>(),
             )
         };
-        let bytes_read = nix::unistd::read(self.file.as_raw_fd(), &mut data_as_buf)?;
-
+        let bytes_read = self.file.read(&mut data_as_buf)?;
         if bytes_read != mem::size_of::<ffi::gpioevent_data>() {
             Ok(None)
         } else {
@@ -1016,7 +1013,7 @@ impl Iterator for LineEventHandle {
         match self.read_event() {
             Ok(None) => None,
             Ok(Some(event)) => Some(Ok(event)),
-            Err(e) => Some(Err(event_err(e))),
+            Err(e) => Some(Err(e.into())),
         }
     }
 }