From 3b154f8049ee47728efb7abd08a3099bc1f97d70 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Fri, 14 May 2021 02:19:22 -0400 Subject: [PATCH 01/12] Implemented most of async pool api --- Cargo.toml | 9 +- examples/read_async.rs | 46 ++++ src/device.rs | 20 +- src/device_handle/async_api.rs | 218 ++++++++++++++++++ .../mod.rs} | 2 + src/device_list.rs | 7 +- src/lib.rs | 8 +- 7 files changed, 290 insertions(+), 20 deletions(-) create mode 100644 examples/read_async.rs create mode 100644 src/device_handle/async_api.rs rename src/{device_handle.rs => device_handle/mod.rs} (99%) diff --git a/Cargo.toml b/Cargo.toml index 1f2396d..deb08a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,10 @@ [package] name = "rusb" version = "0.8.0" -authors = ["David Cuddeback ", "Ilya Averyanov "] +authors = [ + "David Cuddeback ", + "Ilya Averyanov ", +] description = "Rust library for accessing USB devices." license = "MIT" homepage = "https://github.com/a1ien/rusb" @@ -15,7 +18,7 @@ build = "build.rs" travis-ci = { repository = "a1ien/rusb" } [features] -vendored = [ "libusb1-sys/vendored" ] +vendored = ["libusb1-sys/vendored"] [workspace] members = ["libusb1-sys"] @@ -23,6 +26,8 @@ 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" diff --git a/examples/read_async.rs b/examples/read_async.rs new file mode 100644 index 0000000..1a923ea --- /dev/null +++ b/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 = std::env::args().collect(); + + if args.len() < 4 { + eprintln!("Usage: read_async "); + 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) +} diff --git a/src/device.rs b/src/device.rs index bf1a9c3..d703952 100644 --- a/src/device.rs +++ b/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 Device { pub fn get_parent(&self) -> Option { 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, 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 Device { }; Ok(ports[0..ports_number as usize].to_vec()) } - } diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs new file mode 100644 index 0000000..2db06e2 --- /dev/null +++ b/src/device_handle/async_api.rs @@ -0,0 +1,218 @@ +use crate::{DeviceHandle, UsbContext}; +use libusb1_sys as ffi; + +use libc::c_void; +use std::collections::VecDeque; +use std::convert::{TryFrom, TryInto}; +use std::marker::PhantomPinned; +use std::pin::Pin; +use std::ptr::NonNull; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use thiserror::Error; + +type CompletedQueue = Arc>>; +type Transfer = Pin>>; + +#[derive(Error, Debug)] +pub enum AsyncError { + #[error("Transfer timed out")] + TransferTimeout, + #[error("Poll timed out")] + PollTimeout, + #[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), + #[error("Transfer was cancelled")] + Cancelled, +} + +struct AsyncTransfer { + ptr: NonNull, + /// The ID of the transfer, as an idx into the transfer pool + pool_id: usize, + buffer: Vec, + completed_queue: CompletedQueue, + device: Arc>, + // `ptr` holds a pointer to `buffer`, `device`, and `self`, so we must be `!Unpin` + _pin: PhantomPinned, +} +impl AsyncTransfer { + fn new_bulk( + pool_id: usize, + completed_queue: CompletedQueue, + device: Arc>, + endpoint: u8, + buffer: Vec, + timeout: std::time::Duration, + ) -> Pin> { + // 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, + pool_id, + buffer, + completed_queue, + device, + _pin: PhantomPinned, + }); + let user_data = std::ptr::addr_of!(*result) as *mut c_void; + + unsafe { + // Step 2 of async api + ffi::libusb_fill_bulk_transfer( + ptr.as_ptr(), + result.device.as_raw(), + endpoint, + result.buffer.as_ptr() as *mut u8, + result.buffer.capacity().try_into().unwrap(), + Self::transfer_cb, + user_data, // We haven't associated with a queue yet, so this is null + timeout, + ) + }; + result + } + + // 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 + ); + + let transfer: *mut AsyncTransfer = transfer.user_data.cast(); + let transfer = unsafe { &mut *transfer }; + // Mark transfer as completed + transfer + .completed_queue + .lock() + .unwrap() + .push_back(transfer.pool_id); + } + + fn submit(self: &mut Transfer) -> Result<(), AsyncError> { + todo!() + } +} +// TODO: Figure out how to destroy transfers + +/// Represents a pool of asynchronous transfers, that can be polled to completion +pub struct AsyncPool { + /// Contains the pool of AsyncTransfers + pool: Vec>, + /// Contains the idxs of transfers in the `pool` that have completed + completed: CompletedQueue, + device: Arc>, // TODO: We hold refs to this, do we need it to be pinned? +} +impl AsyncPool { + pub fn new_bulk( + device: DeviceHandle, + endpoint: u8, + read_timeout: Duration, + buffers: impl IntoIterator>, + ) -> Result { + let buffers = buffers.into_iter(); + let mut pool = Vec::with_capacity(buffers.size_hint().0); + let completed = Arc::new(Mutex::new(VecDeque::with_capacity(buffers.size_hint().0))); + let device = Arc::new(device); + + for (id, buf) in buffers.into_iter().enumerate() { + let mut transfer: Transfer = AsyncTransfer::new_bulk( + id, + completed.clone(), + device.clone(), + endpoint, + buf, + read_timeout, + ); + transfer.submit()?; + pool.push(transfer); + } + Ok(Self { + pool, + completed, + device, + }) + } + + /// Once a transfer is completed, check the c struct for errors, otherwise swap + /// buffers + fn handle_completed_transfer( + transfer: &mut Transfer, + new_buf: Vec, + ) -> Result, (AsyncError, Vec)> { + todo!() + } + + /// Polls for the completion of async transfers. If successful, will swap the buffer + /// of the completed transfer with `new_buf`, otherwise returns `(err, new_buf)` so + /// that `new_buf` may be repurposed. + pub fn poll( + &mut self, + timeout: Duration, + new_buf: Vec, + ) -> Result, (AsyncError, Vec)> { + let pop_result = { self.completed.lock().unwrap().pop_front() }; + if let Some(id) = pop_result { + let ref mut transfer = self.pool[id]; + return Self::handle_completed_transfer(transfer, new_buf); + } + // No completed transfers, so poll for some new ones + poll_transfers(self.device.context(), timeout); + let pop_result = { self.completed.lock().unwrap().pop_front() }; + if let Some(id) = pop_result { + let ref mut transfer = self.pool[id]; + Self::handle_completed_transfer(transfer, new_buf) + } else { + Err((AsyncError::PollTimeout, new_buf)) + } + } +} + +/// Polls for transfers and executes their callbacks. Will block until the +/// given timeout, or return immediately if timeout is zero. +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() + ), + } + } +} diff --git a/src/device_handle.rs b/src/device_handle/mod.rs similarity index 99% rename from src/device_handle.rs rename to src/device_handle/mod.rs index 767145d..b80873b 100644 --- a/src/device_handle.rs +++ b/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}; diff --git a/src/device_list.rs b/src/device_list.rs index d07630b..3eaf757 100644 --- a/src/device_list.rs +++ b/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 } diff --git a/src/lib.rs b/src/lib.rs index 5dfc530..1b8a28b 100644 --- a/src/lib.rs +++ b/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::AsyncPool, 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), + ) + }) } } From b165387ac8cc65c5a240d0cffa4109a1ec0d58b8 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Tue, 18 May 2021 19:33:16 -0400 Subject: [PATCH 02/12] Updated async example --- examples/read_async.rs | 35 +++++++++++++++++++---------------- 1 file changed, 19 insertions(+), 16 deletions(-) diff --git a/examples/read_async.rs b/examples/read_async.rs index 1a923ea..e75b4d8 100644 --- a/examples/read_async.rs +++ b/examples/read_async.rs @@ -1,4 +1,4 @@ -use rusb::{AsyncTransfer, CbResult, Context, UsbContext}; +use rusb::{AsyncPool, Context, UsbContext}; use std::str::FromStr; use std::time::Duration; @@ -23,24 +23,27 @@ fn main() { const NUM_TRANSFERS: usize = 32; const BUF_SIZE: usize = 1024; - let mut transfers = Vec::new(); + let mut buffers = 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); + let buf = Vec::with_capacity(BUF_SIZE); + buffers.push(buf); } + let mut async_pool = AsyncPool::new_bulk(device, endpoint, Duration::from_secs(5), buffers) + .expect("Failed to create async pool!"); + + let mut swap_vec = Vec::with_capacity(BUF_SIZE); loop { - rusb::poll_transfers(&ctx, Duration::from_secs(10)); + let poll_result = async_pool.poll(Duration::from_secs(10), swap_vec); + match poll_result { + Ok(data) => { + println!("Got data: {:#?}", data); + swap_vec = data + } + Err((err, buf)) => { + eprintln!("Error: {}", err); + swap_vec = buf + } + } } } - -fn callback(result: CbResult) { - println!("{:?}", result) -} From 0117662e65b9803d10759616abdca5ca002bd8c6 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Tue, 18 May 2021 19:44:01 -0400 Subject: [PATCH 03/12] Removed some out of date documentation, cleaned things up --- src/device_handle/async_api.rs | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index 2db06e2..3ec090e 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -88,10 +88,8 @@ impl AsyncTransfer { result } - // 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 + // Part of step 4 of async API the transfer is finished being handled when + // `poll()` is called. 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 }; @@ -112,17 +110,23 @@ impl AsyncTransfer { .push_back(transfer.pool_id); } + // Step 3 of async API fn submit(self: &mut Transfer) -> Result<(), AsyncError> { todo!() } } -// TODO: Figure out how to destroy transfers +impl Drop for AsyncTransfer { + fn drop(&mut self) { + // TODO: Figure out how to destroy transfers, which is step 5 of async API. + todo!() + } +} /// Represents a pool of asynchronous transfers, that can be polled to completion pub struct AsyncPool { /// Contains the pool of AsyncTransfers pool: Vec>, - /// Contains the idxs of transfers in the `pool` that have completed + /// Contains the idxs of transfers in `pool` that have completed completed: CompletedQueue, device: Arc>, // TODO: We hold refs to this, do we need it to be pinned? } @@ -158,7 +162,7 @@ impl AsyncPool { } /// Once a transfer is completed, check the c struct for errors, otherwise swap - /// buffers + /// buffers. Step 4 of async API. fn handle_completed_transfer( transfer: &mut Transfer, new_buf: Vec, From 6089ca7eee8904734bada8fedfb1140e54d3d36b Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Tue, 18 May 2021 21:24:49 -0400 Subject: [PATCH 04/12] Implemented everything but drop strategy --- src/device_handle/async_api.rs | 48 ++++++++++++++++++++++++++++++++-- 1 file changed, 46 insertions(+), 2 deletions(-) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index 3ec090e..564caf2 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -112,7 +112,21 @@ impl AsyncTransfer { // Step 3 of async API fn submit(self: &mut Transfer) -> Result<(), AsyncError> { - todo!() + let transfer_struct = self.ptr; + let errno = unsafe { ffi::libusb_submit_transfer(transfer_struct.as_ptr()) }; + + use ffi::constants::*; + use AsyncError as E; + match errno { + 0 => Ok(()), + LIBUSB_ERROR_NO_DEVICE => Err(E::Disconnected), + LIBUSB_ERROR_BUSY => { + unreachable!("We shouldn't be calling submit on transfers already submitted!") + } + LIBUSB_ERROR_NOT_SUPPORTED => Err(E::Other("Transfer not supported")), + LIBUSB_ERROR_INVALID_PARAM => Err(E::Other("Transfer size bigger than OS supports")), + _ => Err(E::Errno("Error while submitting transfer: ", errno)), + } } } impl Drop for AsyncTransfer { @@ -167,7 +181,37 @@ impl AsyncPool { transfer: &mut Transfer, new_buf: Vec, ) -> Result, (AsyncError, Vec)> { - todo!() + use AsyncError as E; + + let transfer = unsafe { transfer.as_mut().get_unchecked_mut() }; + let transfer_struct = unsafe { transfer.ptr.as_mut() }; + + use ffi::constants::*; + let result = match transfer_struct.status { + LIBUSB_TRANSFER_COMPLETED => Ok(()), + LIBUSB_TRANSFER_CANCELLED => Err(E::Cancelled), + LIBUSB_TRANSFER_ERROR => Err(E::Other("Error occurred during transfer execution")), + LIBUSB_TRANSFER_TIMED_OUT => Err(E::TransferTimeout), + LIBUSB_TRANSFER_STALL => Err(E::Stall), + LIBUSB_TRANSFER_NO_DEVICE => Err(E::Disconnected), + LIBUSB_TRANSFER_OVERFLOW => { + panic!("Device sent more data than expected. Is this even possible when reading?") + } + _ => panic!("Found an unexpected error value for transfer status"), + }; + match result { + Err(err) => Err((err, new_buf)), + Ok(()) => { + debug_assert!(transfer_struct.length >= transfer_struct.actual_length); // sanity + unsafe { + transfer + .buffer + .set_len(transfer_struct.actual_length as usize) + }; + let data = std::mem::replace(&mut transfer.buffer, new_buf); + Ok(data) + } + } } /// Polls for the completion of async transfers. If successful, will swap the buffer From 66b3778c5d6f7ac3f9ce5c4dcda1cc389c7a4518 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Wed, 19 May 2021 13:35:26 -0400 Subject: [PATCH 05/12] Now resubmitting transfers --- src/device_handle/async_api.rs | 33 ++++++++++++++++++++++++++++----- 1 file changed, 28 insertions(+), 5 deletions(-) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index 564caf2..5bd1a63 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -110,8 +110,22 @@ impl AsyncTransfer { .push_back(transfer.pool_id); } + /// Prerequisite: self.buffer ans self.ptr are both correctly set + fn swap_buffer(self: &mut AsyncTransfer, new_buf: Vec) -> Vec { + let transfer_struct = unsafe { self.ptr.as_mut() }; + + let data = std::mem::replace(&mut self.buffer, new_buf); + + // Update transfer struct for new buffer + transfer_struct.actual_length = 0; // TODO: Is this necessary? + transfer_struct.buffer = self.buffer.as_mut_ptr(); + transfer_struct.length = self.buffer.capacity() as i32; + + data + } + // Step 3 of async API - fn submit(self: &mut Transfer) -> Result<(), AsyncError> { + fn submit(self: &mut AsyncTransfer) -> Result<(), AsyncError> { let transfer_struct = self.ptr; let errno = unsafe { ffi::libusb_submit_transfer(transfer_struct.as_ptr()) }; @@ -165,7 +179,7 @@ impl AsyncPool { buf, read_timeout, ); - transfer.submit()?; + unsafe { transfer.as_mut().get_unchecked_mut() }.submit()?; pool.push(transfer); } Ok(Self { @@ -182,7 +196,6 @@ impl AsyncPool { new_buf: Vec, ) -> Result, (AsyncError, Vec)> { use AsyncError as E; - let transfer = unsafe { transfer.as_mut().get_unchecked_mut() }; let transfer_struct = unsafe { transfer.ptr.as_mut() }; @@ -208,8 +221,18 @@ impl AsyncPool { .buffer .set_len(transfer_struct.actual_length as usize) }; - let data = std::mem::replace(&mut transfer.buffer, new_buf); - Ok(data) + + let data = transfer.swap_buffer(new_buf); + + let submit_result = transfer.submit(); + match submit_result { + Ok(()) => Ok(data), + Err(err) => { + // Take back the original buffer and return the error + let new_buf = transfer.swap_buffer(data); + Err((err, new_buf)) + } + } } } } From 65242ead4ccc42a675255b534b35f5b4a39e95fe Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Wed, 19 May 2021 14:09:41 -0400 Subject: [PATCH 06/12] Now using no read timeout (timeout=0) --- examples/read_async.rs | 7 ++++--- src/device_handle/async_api.rs | 24 +++++++----------------- 2 files changed, 11 insertions(+), 20 deletions(-) diff --git a/examples/read_async.rs b/examples/read_async.rs index e75b4d8..38e9940 100644 --- a/examples/read_async.rs +++ b/examples/read_async.rs @@ -29,12 +29,13 @@ fn main() { buffers.push(buf); } - let mut async_pool = AsyncPool::new_bulk(device, endpoint, Duration::from_secs(5), buffers) - .expect("Failed to create async pool!"); + let mut async_pool = + AsyncPool::new_bulk(device, endpoint, buffers).expect("Failed to create async pool!"); let mut swap_vec = Vec::with_capacity(BUF_SIZE); + let timeout = Duration::from_secs(10); loop { - let poll_result = async_pool.poll(Duration::from_secs(10), swap_vec); + let poll_result = async_pool.poll(timeout, swap_vec); match poll_result { Ok(data) => { println!("Got data: {:#?}", data); diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index 5bd1a63..05d874d 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -3,7 +3,7 @@ use libusb1_sys as ffi; use libc::c_void; use std::collections::VecDeque; -use std::convert::{TryFrom, TryInto}; +use std::convert::TryInto; use std::marker::PhantomPinned; use std::pin::Pin; use std::ptr::NonNull; @@ -16,8 +16,6 @@ type Transfer = Pin>>; #[derive(Error, Debug)] pub enum AsyncError { - #[error("Transfer timed out")] - TransferTimeout, #[error("Poll timed out")] PollTimeout, #[error("Transfer is stalled")] @@ -49,14 +47,11 @@ impl AsyncTransfer { device: Arc>, endpoint: u8, buffer: Vec, - timeout: std::time::Duration, ) -> Pin> { // 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 @@ -82,7 +77,7 @@ impl AsyncTransfer { result.buffer.capacity().try_into().unwrap(), Self::transfer_cb, user_data, // We haven't associated with a queue yet, so this is null - timeout, + 0, ) }; result @@ -162,7 +157,6 @@ impl AsyncPool { pub fn new_bulk( device: DeviceHandle, endpoint: u8, - read_timeout: Duration, buffers: impl IntoIterator>, ) -> Result { let buffers = buffers.into_iter(); @@ -171,14 +165,8 @@ impl AsyncPool { let device = Arc::new(device); for (id, buf) in buffers.into_iter().enumerate() { - let mut transfer: Transfer = AsyncTransfer::new_bulk( - id, - completed.clone(), - device.clone(), - endpoint, - buf, - read_timeout, - ); + let mut transfer: Transfer = + AsyncTransfer::new_bulk(id, completed.clone(), device.clone(), endpoint, buf); unsafe { transfer.as_mut().get_unchecked_mut() }.submit()?; pool.push(transfer); } @@ -204,7 +192,9 @@ impl AsyncPool { LIBUSB_TRANSFER_COMPLETED => Ok(()), LIBUSB_TRANSFER_CANCELLED => Err(E::Cancelled), LIBUSB_TRANSFER_ERROR => Err(E::Other("Error occurred during transfer execution")), - LIBUSB_TRANSFER_TIMED_OUT => Err(E::TransferTimeout), + LIBUSB_TRANSFER_TIMED_OUT => { + unreachable!("We are using timeout=0 which means no timeout") + } LIBUSB_TRANSFER_STALL => Err(E::Stall), LIBUSB_TRANSFER_NO_DEVICE => Err(E::Disconnected), LIBUSB_TRANSFER_OVERFLOW => { From af44e5dfbb22076453a978ad7c2a9a75dd3d02b6 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Wed, 19 May 2021 14:16:15 -0400 Subject: [PATCH 07/12] Added AsyncPool.size() --- src/device_handle/async_api.rs | 53 +++++++++++++++++++--------------- 1 file changed, 29 insertions(+), 24 deletions(-) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index 05d874d..fe1d69c 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -177,6 +177,35 @@ impl AsyncPool { }) } + /// Polls for the completion of async transfers. If successful, will swap the buffer + /// of the completed transfer with `new_buf`, otherwise returns `(err, new_buf)` so + /// that `new_buf` may be repurposed. + pub fn poll( + &mut self, + timeout: Duration, + new_buf: Vec, + ) -> Result, (AsyncError, Vec)> { + let pop_result = { self.completed.lock().unwrap().pop_front() }; + if let Some(id) = pop_result { + let ref mut transfer = self.pool[id]; + return Self::handle_completed_transfer(transfer, new_buf); + } + // No completed transfers, so poll for some new ones + poll_transfers(self.device.context(), timeout); + let pop_result = { self.completed.lock().unwrap().pop_front() }; + if let Some(id) = pop_result { + let ref mut transfer = self.pool[id]; + Self::handle_completed_transfer(transfer, new_buf) + } else { + Err((AsyncError::PollTimeout, new_buf)) + } + } + + /// Returns the number of async transfers in the pool. + pub fn size(&self) -> usize { + self.pool.len() + } + /// Once a transfer is completed, check the c struct for errors, otherwise swap /// buffers. Step 4 of async API. fn handle_completed_transfer( @@ -226,30 +255,6 @@ impl AsyncPool { } } } - - /// Polls for the completion of async transfers. If successful, will swap the buffer - /// of the completed transfer with `new_buf`, otherwise returns `(err, new_buf)` so - /// that `new_buf` may be repurposed. - pub fn poll( - &mut self, - timeout: Duration, - new_buf: Vec, - ) -> Result, (AsyncError, Vec)> { - let pop_result = { self.completed.lock().unwrap().pop_front() }; - if let Some(id) = pop_result { - let ref mut transfer = self.pool[id]; - return Self::handle_completed_transfer(transfer, new_buf); - } - // No completed transfers, so poll for some new ones - poll_transfers(self.device.context(), timeout); - let pop_result = { self.completed.lock().unwrap().pop_front() }; - if let Some(id) = pop_result { - let ref mut transfer = self.pool[id]; - Self::handle_completed_transfer(transfer, new_buf) - } else { - Err((AsyncError::PollTimeout, new_buf)) - } - } } /// Polls for transfers and executes their callbacks. Will block until the From e553b67a828cbbb0e424146ec36396b0f811a488 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Wed, 19 May 2021 16:04:30 -0400 Subject: [PATCH 08/12] Renamed Transfer to PinnedTransfer, exported AsyncError --- src/device_handle/async_api.rs | 8 ++++---- src/lib.rs | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index fe1d69c..4732bbf 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -12,7 +12,7 @@ use std::time::Duration; use thiserror::Error; type CompletedQueue = Arc>>; -type Transfer = Pin>>; +type PinnedTransfer = Pin>>; #[derive(Error, Debug)] pub enum AsyncError { @@ -148,7 +148,7 @@ impl Drop for AsyncTransfer { /// Represents a pool of asynchronous transfers, that can be polled to completion pub struct AsyncPool { /// Contains the pool of AsyncTransfers - pool: Vec>, + pool: Vec>, /// Contains the idxs of transfers in `pool` that have completed completed: CompletedQueue, device: Arc>, // TODO: We hold refs to this, do we need it to be pinned? @@ -165,7 +165,7 @@ impl AsyncPool { let device = Arc::new(device); for (id, buf) in buffers.into_iter().enumerate() { - let mut transfer: Transfer = + let mut transfer: PinnedTransfer = AsyncTransfer::new_bulk(id, completed.clone(), device.clone(), endpoint, buf); unsafe { transfer.as_mut().get_unchecked_mut() }.submit()?; pool.push(transfer); @@ -209,7 +209,7 @@ impl AsyncPool { /// Once a transfer is completed, check the c struct for errors, otherwise swap /// buffers. Step 4 of async API. fn handle_completed_transfer( - transfer: &mut Transfer, + transfer: &mut PinnedTransfer, new_buf: Vec, ) -> Result, (AsyncError, Vec)> { use AsyncError as E; diff --git a/src/lib.rs b/src/lib.rs index 1b8a28b..7f56703 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,7 +7,7 @@ pub use crate::{ context::{Context, GlobalContext, Hotplug, LogLevel, Registration, UsbContext}, device::Device, device_descriptor::DeviceDescriptor, - device_handle::async_api::AsyncPool, + device_handle::async_api::{AsyncError, AsyncPool}, device_handle::DeviceHandle, device_list::{DeviceList, Devices}, endpoint_descriptor::EndpointDescriptor, From 781b6c393889761ad0dd7296feee89e6322545bd Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Wed, 19 May 2021 17:00:57 -0400 Subject: [PATCH 09/12] Made AsyncPool Send + Sync --- src/device_handle/async_api.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index 4732bbf..37a42cd 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -256,6 +256,8 @@ impl AsyncPool { } } } +unsafe impl Send for AsyncPool {} +unsafe impl Sync for AsyncPool {} /// Polls for transfers and executes their callbacks. Will block until the /// given timeout, or return immediately if timeout is zero. From 900531e29aa6204e2b95b6beae6b1af0f2bfc1fe Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Fri, 21 May 2021 14:35:17 -0400 Subject: [PATCH 10/12] Implemented transfer cancellation --- src/device_handle/async_api.rs | 38 ++++++++++++++++++++++++++++++++-- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index 37a42cd..aab6bf0 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -95,6 +95,15 @@ impl AsyncTransfer { ffi::constants::LIBUSB_TRANSFER_TYPE_BULK ); + // Before doing anything else, handle the case where the transfer was cancelled + // and we need to clean up the libusb resource + // Step 5 of async API + if transfer.status == ffi::constants::LIBUSB_TRANSFER_CANCELLED { + // NOTE: transfer.user_data has already been deallocated in the destructor. + unsafe { ffi::libusb_free_transfer(transfer) } + return; + } + let transfer: *mut AsyncTransfer = transfer.user_data.cast(); let transfer = unsafe { &mut *transfer }; // Mark transfer as completed @@ -140,8 +149,33 @@ impl AsyncTransfer { } impl Drop for AsyncTransfer { fn drop(&mut self) { - // TODO: Figure out how to destroy transfers, which is step 5 of async API. - todo!() + match unsafe { ffi::libusb_cancel_transfer(self.ptr.as_ptr()) } { + // Doesn't actually cancel until the callback completes, so lets block until + // then + 0 => { + // We already requested the transfer to be cancelled, so this should + // immediately run the callback when libusb_handle_events() is invoked + const TIMEOUT: libc::timeval = libc::timeval { + tv_sec: 0, + tv_usec: 0, + }; + let errno = unsafe { + ffi::libusb_handle_events_timeout_completed( + self.device.context().as_raw(), + &TIMEOUT, + std::ptr::null_mut(), + ) + }; + assert_eq!(errno, 0, "Failed to cancel transfer! ERRNO: {}", errno); + } + // transfer is not in progress, already complete, or already cancelled. + // Therefore, no need to do anything special. + ffi::constants::LIBUSB_ERROR_NOT_FOUND => (), + errno => panic!( + "Error while requesting transfer cancellation! ERRNO: {}", + errno + ), + } } } From c6b98b19f5c2db7bf9063b8417a6b45046667737 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Mon, 24 May 2021 10:04:39 -0400 Subject: [PATCH 11/12] Added overflow error variant --- src/device_handle/async_api.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs index aab6bf0..ba5de60 100644 --- a/src/device_handle/async_api.rs +++ b/src/device_handle/async_api.rs @@ -28,6 +28,8 @@ pub enum AsyncError { Errno(&'static str, i32), #[error("Transfer was cancelled")] Cancelled, + #[error("Transfer could not fit into provided buffer with size of {0}")] + Overflow(usize), } struct AsyncTransfer { @@ -260,9 +262,7 @@ impl AsyncPool { } LIBUSB_TRANSFER_STALL => Err(E::Stall), LIBUSB_TRANSFER_NO_DEVICE => Err(E::Disconnected), - LIBUSB_TRANSFER_OVERFLOW => { - panic!("Device sent more data than expected. Is this even possible when reading?") - } + LIBUSB_TRANSFER_OVERFLOW => Err(E::Overflow(transfer.buffer.capacity())), _ => panic!("Found an unexpected error value for transfer status"), }; match result { From 258fdcf7e44ff26f1c80b52d1a34d897ebee07d3 Mon Sep 17 00:00:00 2001 From: Ryan Butler Date: Mon, 24 May 2021 11:13:10 -0400 Subject: [PATCH 12/12] Made read_async more useful of a program --- examples/read_async.rs | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/examples/read_async.rs b/examples/read_async.rs index 38e9940..4e0cfee 100644 --- a/examples/read_async.rs +++ b/examples/read_async.rs @@ -34,11 +34,15 @@ fn main() { let mut swap_vec = Vec::with_capacity(BUF_SIZE); let timeout = Duration::from_secs(10); + + let mut num_bytes = 0u64; + let mut num_transfers = 0u64; + let mut last_time = std::time::Instant::now(); loop { let poll_result = async_pool.poll(timeout, swap_vec); match poll_result { Ok(data) => { - println!("Got data: {:#?}", data); + num_bytes += data.len() as u64; swap_vec = data } Err((err, buf)) => { @@ -46,5 +50,19 @@ fn main() { swap_vec = buf } } + num_transfers += 1; + + let elapsed = last_time.elapsed(); + if elapsed >= Duration::from_millis(1000) { + let elapsed = elapsed.as_secs_f32(); + println!( + "KiB per second: {}, \tTx per second: {}", + num_bytes as f32 / 1024. / elapsed, + num_transfers as f32 / elapsed + ); + num_bytes = 0; + num_transfers = 0; + last_time = std::time::Instant::now(); + } } }