Merge pull request #125 from kevinmehall/async-rt

Add cargo features for `smol` and `tokio` runtimes

.github/workflows/rust.yml

@@ -21,7 +21,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
-        rust: ['stable', '1.76']
+        rust: ['stable', '1.79']
 
     runs-on: ${{ matrix.os }}
 
@@ -35,7 +35,11 @@ jobs:
     - name: Build
       run: cargo build --verbose
     - name: Run tests
-      run: cargo test --verbose
+      run: |
+        cargo test --verbose
+        cargo test --verbose --features tokio
+        cargo test --verbose --features smol
+        cargo test --verbose --features smol,tokio
 
   build_android:
     runs-on: ubuntu-latest

Cargo.toml

@@ -8,7 +8,7 @@ authors = ["Kevin Mehall <km@kevinmehall.net>"]
 edition = "2021"
 license = "Apache-2.0 OR MIT"
 repository = "https://github.com/kevinmehall/nusb"
-rust-version = "1.76" # keep in sync with .github/workflows/rust.yml
+rust-version = "1.79" # keep in sync with .github/workflows/rust.yml
 
 [dependencies]
 atomic-waker = "1.1.2"
@@ -34,7 +34,16 @@ core-foundation-sys = "0.8.4"
 io-kit-sys = "0.4.0"
 
 [target.'cfg(any(target_os="linux", target_os="android", target_os="windows", target_os="macos"))'.dependencies]
-blocking ="1.6.1"
+blocking = { version = "1.6.1", optional = true }
+tokio = { version = "1", optional = true, features = ["rt"] }
+
+[features]
+# Use the `blocking` crate for making blocking IO async
+smol = ["dep:blocking"]
+
+# Use `tokio`'s IO threadpool for making blocking IO async
+tokio = ["dep:tokio"]
+
 
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(fuzzing)'] }

src/lib.rs

@@ -2,13 +2,11 @@
 //! A new library for cross-platform low-level access to USB devices.
 //!
 //! `nusb` is comparable to the C library [libusb] and its Rust bindings [rusb],
-//! but written in pure Rust. It's built on and exposes async APIs by default,
-//! but can be made blocking using [`futures_lite::future::block_on`][block_on]
-//! or similar.
+//! but written in pure Rust. It supports usage from both async and
+//! blocking contexts, and transfers are natively async.
 //!
 //! [libusb]: https://libusb.info
 //! [rusb]: https://docs.rs/rusb/
-//! [block_on]: https://docs.rs/futures-lite/latest/futures_lite/future/fn.block_on.html
 //!
 //! Use `nusb` to write user-space drivers in Rust for non-standard USB devices
 //! or those without kernel support. For devices implementing a standard USB
@@ -111,8 +109,29 @@
 //!
 //! `nusb` uses IOKit on macOS.
 //!
-//! Users have access to USB devices by default, with no permission configuration needed.
-//! Devices with a kernel driver are not accessible.
+//! Users have access to USB devices by default, with no permission
+//! configuration needed. Devices with a kernel driver are not accessible.
+//!
+//! ## Async support
+//!
+//! Many methods in `nusb` return a [`MaybeFuture`] type, which can either be
+//! `.await`ed (via `IntoFuture`) or `.wait()`ed (blocking the current thread).
+//! This allows for async usage in an async context, or blocking usage in a
+//! non-async context.
+//!
+//! Operations such as [`Device::open`], [`Device::set_configuration`],
+//! [`Device::reset`], [`Device::claim_interface`],
+//! [`Interface::set_alt_setting`], and [`Interface::clear_halt`] require
+//! blocking system calls. To use these in an asynchronous context, `nusb`
+//! relies on an async runtime to run these operations on an IO thread to avoid
+//! blocking in async code. Enable the cargo feature `tokio` or `smol` to use
+//! the corresponding runtime for blocking IO. If neither feature is enabled,
+//! `.await` on these methods will log a warning and block the calling thread.
+//! `.wait()` always runs the blocking operation directly without the overhead
+//! of handing off to an IO thread.
+//!
+//! These features do not affect and are not required for transfers, which are
+//! implemented on top of natively-async OS APIs.
 
 use std::io;
 

src/maybe_future.rs

@@ -8,13 +8,13 @@ use std::{
 ///
 /// A `MaybeFuture` can be run asynchronously with `.await`, or
 /// run synchronously (blocking the current thread) with `.wait()`.
-pub trait MaybeFuture: IntoFuture {
+pub trait MaybeFuture: IntoFuture<IntoFuture: NonWasmSend> + NonWasmSend {
     /// Block waiting for the action to complete
     #[cfg(not(target_arch = "wasm32"))]
     fn wait(self) -> Self::Output;
 
     /// Apply a function to the output.
-    fn map<T: FnOnce(Self::Output) -> R + Unpin, R>(self, f: T) -> Map<Self, T>
+    fn map<T: FnOnce(Self::Output) -> R + Unpin + NonWasmSend, R>(self, f: T) -> Map<Self, T>
     where
         Self: Sized,
     {
@@ -25,6 +25,14 @@ pub trait MaybeFuture: IntoFuture {
     }
 }
 
+#[cfg(not(target_arch = "wasm32"))]
+pub use std::marker::Send as NonWasmSend;
+
+#[cfg(target_arch = "wasm32")]
+pub trait NonWasmSend {}
+#[cfg(target_arch = "wasm32")]
+impl<T> NonWasmSend for T {}
+
 #[cfg(any(
     target_os = "linux",
     target_os = "android",
@@ -33,7 +41,11 @@ pub trait MaybeFuture: IntoFuture {
 ))]
 pub mod blocking {
     use super::MaybeFuture;
-    use std::future::IntoFuture;
+    use std::{
+        future::{Future, IntoFuture},
+        pin::Pin,
+        task::{Context, Poll},
+    };
 
     /// Wrapper that invokes a FnOnce on a background thread when
     /// called asynchronously, or directly when called synchronously.
@@ -54,10 +66,10 @@ pub mod blocking {
     {
         type Output = R;
 
-        type IntoFuture = blocking::Task<R, ()>;
+        type IntoFuture = BlockingTask<R>;
 
         fn into_future(self) -> Self::IntoFuture {
-            blocking::unblock(self.f)
+            BlockingTask::spawn(self.f)
         }
     }
 
@@ -70,6 +82,61 @@ pub mod blocking {
             (self.f)()
         }
     }
+
+    #[cfg(feature = "smol")]
+    pub struct BlockingTask<R>(blocking::Task<R, ()>);
+
+    // If both features are enabled, use `smol` because it does not
+    // require the runtime to be explicitly started
+    #[cfg(all(feature = "tokio", not(feature = "smol")))]
+    pub struct BlockingTask<R>(tokio::task::JoinHandle<R>);
+
+    #[cfg(not(any(feature = "smol", feature = "tokio")))]
+    pub struct BlockingTask<R>(Option<R>);
+
+    impl<R: Send + 'static> BlockingTask<R> {
+        #[cfg(feature = "smol")]
+        fn spawn(f: impl FnOnce() -> R + Send + 'static) -> Self {
+            Self(blocking::unblock(f))
+        }
+
+        #[cfg(all(feature = "tokio", not(feature = "smol")))]
+        fn spawn(f: impl FnOnce() -> R + Send + 'static) -> Self {
+            Self(tokio::task::spawn_blocking(f))
+        }
+
+        #[cfg(not(any(feature = "smol", feature = "tokio")))]
+        fn spawn(f: impl FnOnce() -> R + Send + 'static) -> Self {
+            static ONCE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true);
+
+            if ONCE.swap(false, std::sync::atomic::Ordering::Relaxed) {
+                log::warn!("Awaiting blocking syscall without an async runtime: enable the `smol` or `tokio` feature of `nusb` to avoid blocking the thread.")
+            }
+
+            Self(Some(f()))
+        }
+    }
+
+    impl<R> Unpin for BlockingTask<R> {}
+
+    impl<R> Future for BlockingTask<R> {
+        type Output = R;
+
+        #[cfg(feature = "smol")]
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            Pin::new(&mut self.0).poll(cx)
+        }
+
+        #[cfg(all(feature = "tokio", not(feature = "smol")))]
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+            Pin::new(&mut self.0).poll(cx).map(|r| r.unwrap())
+        }
+
+        #[cfg(not(any(feature = "smol", feature = "tokio")))]
+        fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+            Poll::Ready(self.0.take().expect("polled after completion"))
+        }
+    }
 }
 
 pub(crate) struct Ready<T>(pub(crate) T);
@@ -83,7 +150,8 @@ impl<T> IntoFuture for Ready<T> {
     }
 }
 
-impl<T> MaybeFuture for Ready<T> {
+impl<T: NonWasmSend> MaybeFuture for Ready<T> {
+    #[cfg(not(target_arch = "wasm32"))]
     fn wait(self) -> Self::Output {
         self.0
     }
@@ -106,7 +174,8 @@ impl<F: MaybeFuture, T: FnOnce(F::Output) -> R, R> IntoFuture for Map<F, T> {
     }
 }
 
-impl<F: MaybeFuture, T: FnOnce(F::Output) -> R, R> MaybeFuture for Map<F, T> {
+impl<F: MaybeFuture, T: FnOnce(F::Output) -> R + NonWasmSend, R> MaybeFuture for Map<F, T> {
+    #[cfg(not(target_arch = "wasm32"))]
     fn wait(self) -> Self::Output {
         (self.func)(self.wrapped.wait())
     }