Add Bulk transfer async API

Cargo.toml

@@ -1,7 +1,10 @@
 [package]
 name = "rusb"
 version = "0.8.0"
-authors = ["David Cuddeback <[email protected]>", "Ilya Averyanov <[email protected]>"]
+authors = [
+    "David Cuddeback <[email protected]>",
+    "Ilya Averyanov <[email protected]>",
+]
 description = "Rust library for accessing USB devices."
 license = "MIT"
 homepage = "https://github.com/a1ien/rusb"
@@ -15,14 +18,16 @@ build = "build.rs"
 travis-ci = { repository = "a1ien/rusb" }
 
 [features]
-vendored = [ "libusb1-sys/vendored" ]
+vendored = ["libusb1-sys/vendored"]
 
 [workspace]
 members = ["libusb1-sys"]
 
 [dependencies]
 libusb1-sys = { path = "libusb1-sys", version = "0.5.0" }
 libc = "0.2"
+log = "0.4"
+thiserror = "1"
 
 [dev-dependencies]
 regex = "1"

examples/read_async.rs

@@ -0,0 +1,46 @@
+use rusb::{AsyncTransfer, CbResult, Context, UsbContext};
+
+use std::str::FromStr;
+use std::time::Duration;
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+
+    if args.len() < 4 {
+        eprintln!("Usage: read_async <vendor-id> <product-id> <endpoint>");
+        return;
+    }
+
+    let vid: u16 = FromStr::from_str(args[1].as_ref()).unwrap();
+    let pid: u16 = FromStr::from_str(args[2].as_ref()).unwrap();
+    let endpoint: u8 = FromStr::from_str(args[3].as_ref()).unwrap();
+
+    let ctx = Context::new().expect("Could not initialize libusb");
+    let device = ctx
+        .open_device_with_vid_pid(vid, pid)
+        .expect("Could not find device");
+
+    const NUM_TRANSFERS: usize = 32;
+    const BUF_SIZE: usize = 1024;
+
+    let mut transfers = Vec::new();
+    for _ in 0..NUM_TRANSFERS {
+        let mut transfer = AsyncTransfer::new_bulk(
+            &device,
+            endpoint,
+            BUF_SIZE,
+            callback,
+            Duration::from_secs(10),
+        );
+        transfer.submit().expect("Could not submit transfer");
+        transfers.push(transfer);
+    }
+
+    loop {
+        rusb::poll_transfers(&ctx, Duration::from_secs(10));
+    }
+}
+
+fn callback(result: CbResult) {
+    println!("{:?}", result)
+}

src/device.rs

@@ -10,9 +10,9 @@ use crate::{
     config_descriptor::{self, ConfigDescriptor},
     device_descriptor::{self, DeviceDescriptor},
     device_handle::DeviceHandle,
+    error,
     fields::{self, Speed},
     Error, UsbContext,
-    error,
 };
 
 /// A reference to a USB device.
@@ -150,27 +150,16 @@ impl<T: UsbContext> Device<T> {
     pub fn get_parent(&self) -> Option<Self> {
         let device = unsafe { libusb_get_parent(self.device.as_ptr()) };
         NonNull::new(device)
-            .map(|device|
-                 unsafe {
-                     Device::from_libusb(
-                         self.context.clone(),
-                         device,
-                     )
-                 }
-            )
+            .map(|device| unsafe { Device::from_libusb(self.context.clone(), device) })
     }
 
     ///  Get the list of all port numbers from root for the specified device
     pub fn port_numbers(&self) -> Result<Vec<u8>, Error> {
         // As per the USB 3.0 specs, the current maximum limit for the depth is 7.
-        let mut ports = [0;7];
+        let mut ports = [0; 7];
 
         let result = unsafe {
-            libusb_get_port_numbers(
-                self.device.as_ptr(),
-                ports.as_mut_ptr(),
-                ports.len() as i32
-            )
+            libusb_get_port_numbers(self.device.as_ptr(), ports.as_mut_ptr(), ports.len() as i32)
         };
 
         let ports_number = if result < 0 {
@@ -180,5 +169,4 @@ impl<T: UsbContext> Device<T> {
         };
         Ok(ports[0..ports_number as usize].to_vec())
     }
-
 }

src/device_handle/async_api.rs

@@ -0,0 +1,218 @@
+use crate::{DeviceHandle, UsbContext};
+
+use libc::c_void;
+use libusb1_sys as ffi;
+use thiserror::Error;
+
+use std::convert::{TryFrom, TryInto};
+use std::marker::{PhantomData, PhantomPinned};
+use std::pin::Pin;
+use std::ptr::NonNull;
+use std::time::Duration;
+
+pub type CbResult<'a> = Result<&'a [u8], TransferError>;
+
+#[derive(Error, Debug)]
+pub enum TransferError {
+    #[error("Transfer timed out")]
+    Timeout,
+    #[error("Transfer is stalled")]
+    Stall,
+    #[error("Device was disconnected")]
+    Disconnected,
+    #[error("Other Error: {0}")]
+    Other(&'static str),
+    #[error("{0}ERRNO: {1}")]
+    Errno(&'static str, i32),
+}
+
+pub struct AsyncTransfer<'d, C: UsbContext, F> {
+    ptr: NonNull<ffi::libusb_transfer>,
+    closure: F,
+    buffer: Box<[u8]>,
+    _pin: PhantomPinned, // `ptr` holds a ptr to `closure`, so mark !Unpin
+    _device: PhantomData<&'d DeviceHandle<C>>,
+}
+impl<'d, 'b, C: 'd + UsbContext, F: FnMut(CbResult<'b>) + Send> AsyncTransfer<'d, C, F> {
+    #[allow(unused)]
+    pub fn new_bulk(
+        device: &'d DeviceHandle<C>,
+        endpoint: u8,
+        buf_size: usize,
+        callback: F,
+        timeout: std::time::Duration,
+    ) -> Pin<Box<Self>> {
+        // non-isochronous endpoints (e.g. control, bulk, interrupt) specify a value of 0
+        // This is step 1 of async API
+        let ptr = unsafe { ffi::libusb_alloc_transfer(0) };
+        let ptr = NonNull::new(ptr).expect("Could not allocate transfer!");
+        let timeout = libc::c_uint::try_from(timeout.as_millis())
+            .expect("Duration was too long to fit into a c_uint");
+
+        // Safety: Pinning `result` ensures it doesn't move, but we know that we will
+        // want to access its fields mutably, we just don't want its memory location
+        // changing (or its fields moving!). So routinely we will unsafely interact with
+        // its fields mutably through a shared reference, but this is still sound.
+        let result = Box::pin(Self {
+            ptr,
+            closure: callback,
+            buffer: vec![0u8; buf_size].into_boxed_slice(),
+            _pin: PhantomPinned,
+            _device: PhantomData,
+        });
+
+        unsafe {
+            // This casting, and passing it to the transfer struct, relies on
+            // the pointer being a regular pointer and not a fat pointer.
+            // Also, closure will be invoked from whatever thread polls, which
+            // may be different from the current thread. So it must be `Send`.
+            // Also, although many threads at once may poll concurrently, only
+            // one will actually ever execute the transfer at a time, so we do
+            // not need to worry about simultaneous writes to the buffer
+            let closure_as_ptr: *mut F = {
+                let ptr: *const F = &result.closure;
+                ptr as *mut F
+            };
+            // Step 2 of async api
+            ffi::libusb_fill_bulk_transfer(
+                ptr.as_ptr(),
+                device.as_raw(),
+                endpoint,
+                result.buffer.as_ptr() as *mut u8,
+                result.buffer.len().try_into().unwrap(),
+                Self::transfer_cb,
+                closure_as_ptr.cast(),
+                timeout,
+            )
+        };
+        result
+    }
+
+    /// Submits a transfer to libusb's event engine.
+    /// # Panics
+    /// A transfer should not be double-submitted! Only re-submit after a submission has
+    /// returned Err, or the callback has gotten an Err.
+    // Step 3 of async API
+    #[allow(unused)]
+    pub fn submit(self: &mut Pin<Box<Self>>) -> Result<(), TransferError> {
+        let errno = unsafe { ffi::libusb_submit_transfer(self.ptr.as_ptr()) };
+        use ffi::constants::*;
+        match errno {
+            0 => Ok(()),
+            LIBUSB_ERROR_BUSY => {
+                panic!("Do not double-submit a transfer!")
+            }
+            LIBUSB_ERROR_NOT_SUPPORTED => Err(TransferError::Other("Unsupported transfer!")),
+            LIBUSB_ERROR_INVALID_PARAM => Err(TransferError::Other("Transfer size too large!")),
+            LIBUSB_ERROR_NO_DEVICE => Err(TransferError::Disconnected),
+            _ => Err(TransferError::Errno("Unable to submit transfer. ", errno)),
+        }
+    }
+
+    // We need to invoke our closure using a c-style function, so we store the closure
+    // inside the custom user data field of the transfer struct, and then call the
+    // user provided closure from there.
+    // Step 4 of async API
+    extern "system" fn transfer_cb(transfer: *mut ffi::libusb_transfer) {
+        // Safety: libusb should never make this null, so this is fine
+        let transfer = unsafe { &mut *transfer };
+
+        // sanity
+        debug_assert_eq!(
+            transfer.transfer_type,
+            ffi::constants::LIBUSB_TRANSFER_TYPE_BULK
+        );
+
+        // sanity
+        debug_assert_eq!(
+            std::mem::size_of::<*mut F>(),
+            std::mem::size_of::<*mut c_void>(),
+        );
+        // Safety: The pointer shouldn't be a fat pointer, and should be valid, so
+        // this should be sound
+        let closure = unsafe {
+            let closure: *mut F = std::mem::transmute(transfer.user_data);
+            &mut *closure
+        };
+
+        use ffi::constants::*;
+        match transfer.status {
+            LIBUSB_TRANSFER_CANCELLED => {
+                // Step 5 of async API: Transfer was cancelled, free the transfer
+                unsafe { ffi::libusb_free_transfer(transfer) }
+            }
+            LIBUSB_TRANSFER_COMPLETED => {
+                debug_assert!(transfer.length >= transfer.actual_length); // sanity
+                let data = unsafe {
+                    std::slice::from_raw_parts(transfer.buffer, transfer.actual_length as usize)
+                };
+                (*closure)(Ok(data));
+            }
+            LIBUSB_TRANSFER_ERROR => (*closure)(Err(TransferError::Other(
+                "Error occurred during transfer execution",
+            ))),
+            LIBUSB_TRANSFER_TIMED_OUT => {
+                (*closure)(Err(TransferError::Timeout));
+            }
+            LIBUSB_TRANSFER_STALL => (*closure)(Err(TransferError::Stall)),
+            LIBUSB_TRANSFER_NO_DEVICE => (*closure)(Err(TransferError::Disconnected)),
+            LIBUSB_TRANSFER_OVERFLOW => unreachable!(),
+            _ => panic!("Found an unexpected error value for transfer status"),
+        }
+    }
+}
+impl<C: UsbContext, F> AsyncTransfer<'_, C, F> {
+    /// Helper function for the Drop impl.
+    fn drop_helper(self: Pin<&mut Self>) {
+        // Actual drop code goes here.
+        let transfer_ptr = self.ptr.as_ptr();
+        let errno = unsafe { ffi::libusb_cancel_transfer(transfer_ptr) };
+        match errno {
+            0 | ffi::constants::LIBUSB_ERROR_NOT_FOUND => (),
+            errno => {
+                log::warn!(
+                    "Could not cancel USB transfer. Memory may be leaked. Errno: {}, Error message: {}",
+                    errno, unsafe{std::ffi::CStr::from_ptr( ffi::libusb_strerror(errno))}.to_str().unwrap()
+                )
+            }
+        }
+    }
+}
+
+impl<C: UsbContext, F> Drop for AsyncTransfer<'_, C, F> {
+    fn drop(&mut self) {
+        // We call `drop_helper` because that function represents the actual semantics
+        // that `self` has when being dropped.
+        // (see https://doc.rust-lang.org/std/pin/index.html#drop-implementation)
+        // Safety: `new_unchecked` is okay because we know this value is never used
+        // again after being dropped.
+        Self::drop_helper(unsafe { Pin::new_unchecked(self) });
+    }
+}
+
+/// Polls for transfers and executes their callbacks. Will block until the
+/// given timeout, or return immediately if timeout is zero.
+/// Returns whether a transfer was completed
+pub fn poll_transfers(ctx: &impl UsbContext, timeout: Duration) {
+    let timeval = libc::timeval {
+        tv_sec: timeout.as_secs().try_into().unwrap(),
+        tv_usec: timeout.subsec_millis().try_into().unwrap(),
+    };
+    unsafe {
+        let errno = ffi::libusb_handle_events_timeout_completed(
+            ctx.as_raw(),
+            std::ptr::addr_of!(timeval),
+            std::ptr::null_mut(),
+        );
+        use ffi::constants::*;
+        match errno {
+            0 => (),
+            LIBUSB_ERROR_INVALID_PARAM => panic!("Provided timeout was unexpectedly invalid"),
+            _ => panic!(
+                "Error when polling transfers. ERRNO: {}, Message: {}",
+                errno,
+                std::ffi::CStr::from_ptr(ffi::libusb_strerror(errno)).to_string_lossy()
+            ),
+        }
+    }
+}

src/device_handle.rs → src/device_handle/mod.rs

@@ -1,3 +1,5 @@
+pub mod async_api;
+
 use std::{mem, ptr::NonNull, time::Duration, u8};
 
 use libc::{c_int, c_uchar, c_uint};

src/device_list.rs

@@ -102,7 +102,12 @@ impl<'a, T: UsbContext> Iterator for Devices<'a, T> {
             let device = self.devices[self.index];
 
             self.index += 1;
-            Some(unsafe { device::Device::from_libusb(self.context.clone(), std::ptr::NonNull::new_unchecked(device)) })
+            Some(unsafe {
+                device::Device::from_libusb(
+                    self.context.clone(),
+                    std::ptr::NonNull::new_unchecked(device),
+                )
+            })
         } else {
             None
         }

src/lib.rs

@@ -7,6 +7,7 @@ pub use crate::{
     context::{Context, GlobalContext, Hotplug, LogLevel, Registration, UsbContext},
     device::Device,
     device_descriptor::DeviceDescriptor,
+    device_handle::async_api::{poll_transfers, AsyncTransfer, CbResult},
     device_handle::DeviceHandle,
     device_list::{DeviceList, Devices},
     endpoint_descriptor::EndpointDescriptor,
@@ -106,6 +107,11 @@ pub fn open_device_with_vid_pid(
     if handle.is_null() {
         None
     } else {
-        Some(unsafe { DeviceHandle::from_libusb(GlobalContext::default(), std::ptr::NonNull::new_unchecked(handle)) })
+        Some(unsafe {
+            DeviceHandle::from_libusb(
+                GlobalContext::default(),
+                std::ptr::NonNull::new_unchecked(handle),
+            )
+        })
     }
 }