From c9d3705ff7dc4385f702c94f0b716dd4d0d853e3 Mon Sep 17 00:00:00 2001 From: thegecko Date: Fri, 11 Sep 2026 18:09:44 -0700 Subject: [PATCH 1/2] Fix concurrent access of endpoints --- src/webusb_device.rs | 106 ++++++++++++++++++++++++++++++++++++++----- test/webusb.js | 31 +++++++++++++ 2 files changed, 125 insertions(+), 12 deletions(-) diff --git a/src/webusb_device.rs b/src/webusb_device.rs index 7dc37af..73709d9 100644 --- a/src/webusb_device.rs +++ b/src/webusb_device.rs @@ -4,7 +4,7 @@ use nusb::{ descriptors::language_id::US_ENGLISH, descriptors::TransferType, transfer::Buffer, transfer::Bulk, transfer::Interrupt, MaybeFuture, }; -use std::time::Duration; +use std::{collections::HashMap, sync::Arc, sync::Mutex, sync::MutexGuard, time::Duration}; const ENDPOINT_NUMBER_MASK: u8 = 0x7f; const DESC_TIMEOUT: Duration = Duration::from_millis(100); @@ -99,6 +99,48 @@ enum AnyEndpoint { Interrupt(nusb::Endpoint), } +struct SharedEndpointCache { + endpoints: Mutex>>>, +} + +impl SharedEndpointCache { + fn new() -> Self { + Self { + endpoints: Mutex::new(HashMap::new()), + } + } + + fn guard(&self) -> MutexGuard<'_, HashMap>>> { + self.endpoints + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn clear(&self) { + self.guard().clear(); + } + + fn get_or_try_insert_with(&self, endpointNumber: u8, open: F) -> Option>> + where + F: FnOnce() -> Option, + { + let mut endpoints = self.guard(); + if let Some(endpoint) = endpoints.get(&endpointNumber) { + return Some(endpoint.clone()); + } + + let endpoint = Arc::new(Mutex::new(open()?)); + endpoints.insert(endpointNumber, endpoint.clone()); + Some(endpoint) + } +} + +fn endpoint_guard(endpoint: &Mutex) -> MutexGuard<'_, T> { + endpoint + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + impl AnyEndpoint { fn max_packet_size(&self) -> usize { match self { @@ -268,6 +310,8 @@ pub struct UsbDevice { device_info: nusb::DeviceInfo, device: Option, interfaces: Vec>, + in_endpoints: SharedEndpointCache>, + out_endpoints: SharedEndpointCache>, #[napi(writable = false)] pub vendorId: u16, @@ -315,6 +359,8 @@ impl UsbDevice { device_info: device_info.clone(), device: None, interfaces: vec![None; 256], + in_endpoints: SharedEndpointCache::new(), + out_endpoints: SharedEndpointCache::new(), vendorId: device_info.vendor_id(), productId: device_info.product_id(), deviceVersionMajor, @@ -440,6 +486,7 @@ impl UsbDevice { #[napi] pub async unsafe fn open(&mut self) -> Result<()> { + self.clear_endpoint_caches(); let device_info = self.device_info.clone(); let device = run_blocking(move || { device_info @@ -454,6 +501,7 @@ impl UsbDevice { #[napi] pub async unsafe fn close(&mut self) -> Result<()> { + self.clear_endpoint_caches(); self.device = None; Ok(()) } @@ -467,6 +515,7 @@ impl UsbDevice { pub async fn reset(&self) -> Result<()> { match &self.device { Some(device) => { + self.clear_endpoint_caches(); let device = device.clone(); run_blocking(move || { device @@ -474,7 +523,9 @@ impl UsbDevice { .wait() .map_err(|e| format!("reset error: {e}")) }) - .await + .await?; + self.clear_endpoint_caches(); + Ok(()) } None => Err(napi::Error::from_reason("reset error: invalid state")), } @@ -484,6 +535,7 @@ impl UsbDevice { pub async fn selectConfiguration(&self, configurationValue: u8) -> Result<()> { match &self.device { Some(device) => { + self.clear_endpoint_caches(); let found = device .configurations() .any(|c| c.configuration_value() == configurationValue); @@ -507,7 +559,9 @@ impl UsbDevice { .wait() .map_err(|e| format!("selectConfiguration error: {e}")) }) - .await + .await?; + self.clear_endpoint_caches(); + Ok(()) } } None => Err(napi::Error::from_reason( @@ -529,6 +583,7 @@ impl UsbDevice { }) .await?; self.interfaces[interfaceNumber as usize] = Some(interface); + self.clear_endpoint_caches(); Ok(()) } None => Err(napi::Error::from_reason( @@ -542,6 +597,7 @@ impl UsbDevice { match &self.device { Some(_device) => match &self.interfaces[interfaceNumber as usize] { Some(_interface) => { + self.clear_endpoint_caches(); self.interfaces[interfaceNumber as usize] = None; Ok(()) } @@ -563,6 +619,7 @@ impl UsbDevice { ) -> Result<()> { match &self.interfaces[interfaceNumber as usize] { Some(interface) => { + self.clear_endpoint_caches(); let interface = interface.clone(); run_blocking(move || { interface @@ -669,9 +726,10 @@ impl UsbDevice { timeout: u32, length: u32, ) -> Result> { - match self.get_endpoint::(endpointNumber) { - Some(mut endpoint) => { + match self.get_in_endpoint(endpointNumber) { + Some(endpoint) => { let v = run_blocking(move || { + let mut endpoint = endpoint_guard(&endpoint); let packet_size = endpoint.max_packet_size(); let req = (((length as usize) + packet_size - 1) / packet_size) * packet_size; let buf = Buffer::new(req); @@ -702,10 +760,11 @@ impl UsbDevice { timeout: u32, data: Uint8Array, ) -> Result { - match self.get_endpoint::(endpointNumber) { - Some(mut endpoint) => { + match self.get_out_endpoint(endpointNumber) { + Some(endpoint) => { let data = data.to_vec(); run_blocking(move || { + let mut endpoint = endpoint_guard(&endpoint); let mut buf = Buffer::new(data.len()); buf.extend_from_slice(&data); let completion = @@ -763,9 +822,10 @@ impl UsbDevice { endpointNumber: u8, ) -> Result<()> { if direction == "in" { - match self.get_endpoint::(endpointNumber) { - Some(mut endpoint) => { + match self.get_in_endpoint(endpointNumber) { + Some(endpoint) => { run_blocking(move || { + let mut endpoint = endpoint_guard(&endpoint); endpoint .clear_halt_blocking() .map_err(|e| format!("clearHalt error: {e}")) @@ -779,9 +839,10 @@ impl UsbDevice { } } } else { - match self.get_endpoint::(endpointNumber) { - Some(mut endpoint) => { + match self.get_out_endpoint(endpointNumber) { + Some(endpoint) => { run_blocking(move || { + let mut endpoint = endpoint_guard(&endpoint); endpoint .clear_halt_blocking() .map_err(|e| format!("clearHalt error: {e}")) @@ -799,6 +860,11 @@ impl UsbDevice { Ok(()) } + fn clear_endpoint_caches(&self) { + self.in_endpoints.clear(); + self.out_endpoints.clear(); + } + #[napi] pub async fn detachKernelDriver(&self, interfaceNumber: u8) -> Result<()> { match &self.device { @@ -878,7 +944,23 @@ impl UsbDevice { None } - fn get_endpoint( + fn get_in_endpoint( + &self, + endpointNumber: u8, + ) -> Option>>> { + self.in_endpoints + .get_or_try_insert_with(endpointNumber, || self.open_endpoint(endpointNumber)) + } + + fn get_out_endpoint( + &self, + endpointNumber: u8, + ) -> Option>>> { + self.out_endpoints + .get_or_try_insert_with(endpointNumber, || self.open_endpoint(endpointNumber)) + } + + fn open_endpoint( &self, endpointNumber: u8, ) -> Option> { diff --git a/test/webusb.js b/test/webusb.js index 559756d..2a1d130 100644 --- a/test/webusb.js +++ b/test/webusb.js @@ -311,6 +311,20 @@ describe('Transfers', () => { assert.equal(transferResult.bytesWritten, b2.byteLength); }); + it('allows concurrent transferOut calls to the same endpoint', async () => { + const transfers = [ + device.transferOut(4, b2), + device.transferOut(4, b2), + ]; + const results = await Promise.allSettled(transfers); + + assert.deepEqual(results.map(result => result.status), ['fulfilled', 'fulfilled']); + for (const result of results) { + assert.equal(result.value.status, 'ok'); + assert.equal(result.value.bytesWritten, b2.byteLength); + } + }); + it('should transfer IN', async () => { const transferResult = await device.transferIn(3, b2.byteLength); @@ -322,6 +336,23 @@ describe('Transfers', () => { assert(resultBuffer.equals(expectedBuffer)); }); + it('allows concurrent transferIn calls to the same endpoint', async () => { + await device.transferOut(4, b2); + await device.transferOut(4, b2); + + const transfers = [ + device.transferIn(3, b2.byteLength), + device.transferIn(3, b2.byteLength), + ]; + const results = await Promise.allSettled(transfers); + + assert.deepEqual(results.map(result => result.status), ['fulfilled', 'fulfilled']); + for (const result of results) { + assert.equal(result.value.status, 'ok'); + assert.equal(result.value.data.byteLength, b2.byteLength); + } + }); + after(async () => { await device.releaseInterface(0); await device.close(); From 5884bf56761a340404ec82f4d971ad7e393eed7f Mon Sep 17 00:00:00 2001 From: thegecko Date: Fri, 11 Sep 2026 18:22:50 -0700 Subject: [PATCH 2/2] Use multiple transfers in nusb queue --- src/webusb_device.rs | 409 ++++++++++++++++++++++++++++++++++--------- 1 file changed, 330 insertions(+), 79 deletions(-) diff --git a/src/webusb_device.rs b/src/webusb_device.rs index 73709d9..0511527 100644 --- a/src/webusb_device.rs +++ b/src/webusb_device.rs @@ -4,7 +4,13 @@ use nusb::{ descriptors::language_id::US_ENGLISH, descriptors::TransferType, transfer::Buffer, transfer::Bulk, transfer::Interrupt, MaybeFuture, }; -use std::{collections::HashMap, sync::Arc, sync::Mutex, sync::MutexGuard, time::Duration}; +use std::{ + collections::{HashMap, VecDeque}, + marker::PhantomData, + sync::{mpsc, Arc, Mutex, MutexGuard}, + thread::{self, JoinHandle}, + time::{Duration, Instant}, +}; const ENDPOINT_NUMBER_MASK: u8 = 0x7f; const DESC_TIMEOUT: Duration = Duration::from_millis(100); @@ -100,7 +106,7 @@ enum AnyEndpoint { } struct SharedEndpointCache { - endpoints: Mutex>>>, + endpoints: Mutex>, } impl SharedEndpointCache { @@ -110,37 +116,43 @@ impl SharedEndpointCache { } } - fn guard(&self) -> MutexGuard<'_, HashMap>>> { + fn guard(&self) -> MutexGuard<'_, HashMap> { self.endpoints .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) } +} +impl SharedEndpointCache> { fn clear(&self) { - self.guard().clear(); + let workers = { + let mut endpoints = self.guard(); + endpoints + .drain() + .map(|(_, worker)| worker) + .collect::>() + }; + + for worker in workers { + worker.stop(); + } } - fn get_or_try_insert_with(&self, endpointNumber: u8, open: F) -> Option>> + fn get_or_try_insert_with(&self, endpointNumber: u8, open: F) -> Option> where - F: FnOnce() -> Option, + F: FnOnce() -> Option>, { let mut endpoints = self.guard(); if let Some(endpoint) = endpoints.get(&endpointNumber) { return Some(endpoint.clone()); } - let endpoint = Arc::new(Mutex::new(open()?)); - endpoints.insert(endpointNumber, endpoint.clone()); - Some(endpoint) + let worker = EndpointWorker::new(open()?); + endpoints.insert(endpointNumber, worker.clone()); + Some(worker) } } -fn endpoint_guard(endpoint: &Mutex) -> MutexGuard<'_, T> { - endpoint - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) -} - impl AnyEndpoint { fn max_packet_size(&self) -> usize { match self { @@ -149,14 +161,31 @@ impl AnyEndpoint { } } - fn transfer_blocking( - &mut self, - buf: nusb::transfer::Buffer, - timeout: Duration, - ) -> nusb::transfer::Completion { + fn submit(&mut self, buf: nusb::transfer::Buffer) { match self { - AnyEndpoint::Bulk(ep) => ep.transfer_blocking(buf, timeout), - AnyEndpoint::Interrupt(ep) => ep.transfer_blocking(buf, timeout), + AnyEndpoint::Bulk(ep) => ep.submit(buf), + AnyEndpoint::Interrupt(ep) => ep.submit(buf), + } + } + + fn pending(&self) -> usize { + match self { + AnyEndpoint::Bulk(ep) => ep.pending(), + AnyEndpoint::Interrupt(ep) => ep.pending(), + } + } + + fn wait_next_complete(&mut self, timeout: Duration) -> Option { + match self { + AnyEndpoint::Bulk(ep) => ep.wait_next_complete(timeout), + AnyEndpoint::Interrupt(ep) => ep.wait_next_complete(timeout), + } + } + + fn cancel_all(&mut self) { + match self { + AnyEndpoint::Bulk(ep) => ep.cancel_all(), + AnyEndpoint::Interrupt(ep) => ep.cancel_all(), } } @@ -168,6 +197,249 @@ impl AnyEndpoint { } } +type TransferResponse = mpsc::Sender>; +type ClearHaltResponse = mpsc::Sender>; + +enum EndpointCommand { + Transfer { + buffer: Buffer, + timeout: Duration, + response: TransferResponse, + }, + ClearHalt { + response: ClearHaltResponse, + }, + Stop, +} + +struct PendingTransfer { + deadline: Instant, + response: Option, +} + +struct EndpointWorker { + sender: mpsc::Sender, + join_handle: Arc>>>, + max_packet_size: usize, + _direction: PhantomData, +} + +impl Clone for EndpointWorker { + fn clone(&self) -> Self { + Self { + sender: self.sender.clone(), + join_handle: self.join_handle.clone(), + max_packet_size: self.max_packet_size, + _direction: PhantomData, + } + } +} + +impl EndpointWorker { + fn new(endpoint: AnyEndpoint) -> Self { + const ENDPOINT_POLL_TIMEOUT: Duration = Duration::from_millis(10); + + let max_packet_size = endpoint.max_packet_size(); + let (sender, receiver) = mpsc::channel(); + let join_handle = thread::spawn(move || { + let mut endpoint = endpoint; + let mut pending = VecDeque::new(); + let mut clear_halt_responses: Vec = Vec::new(); + let mut stopping = false; + + loop { + if pending.is_empty() && stopping { + break; + } + + if pending.is_empty() { + match receiver.recv() { + Ok(command) => handle_endpoint_command( + command, + &mut endpoint, + &mut pending, + &mut clear_halt_responses, + &mut stopping, + ), + Err(_) => break, + } + } + + loop { + match receiver.try_recv() { + Ok(command) => handle_endpoint_command( + command, + &mut endpoint, + &mut pending, + &mut clear_halt_responses, + &mut stopping, + ), + Err(mpsc::TryRecvError::Empty) => break, + Err(mpsc::TryRecvError::Disconnected) => { + stopping = true; + break; + } + } + } + + expire_timed_out_transfers(&mut pending); + + if endpoint.pending() > 0 { + if let Some(completion) = endpoint.wait_next_complete(next_wait(&pending)) { + if let Some(mut pending_transfer) = pending.pop_front() { + if let Some(response) = pending_transfer.response.take() { + let _ = response.send(Ok(completion)); + } + } + } + } else if !clear_halt_responses.is_empty() { + let result = endpoint.clear_halt_blocking().map_err(|e| format!("{e}")); + for response in clear_halt_responses.drain(..) { + let _ = response.send(result.clone()); + } + } else if !pending.is_empty() { + thread::sleep(ENDPOINT_POLL_TIMEOUT); + } + } + }); + + Self { + sender, + join_handle: Arc::new(Mutex::new(Some(join_handle))), + max_packet_size, + _direction: PhantomData, + } + } + + async fn transfer( + &self, + buffer: Buffer, + timeout: Duration, + ) -> Result { + let (response, receiver) = mpsc::channel(); + self.sender + .send(EndpointCommand::Transfer { + buffer, + timeout, + response, + }) + .map_err(|_| napi::Error::from_reason("endpoint worker stopped"))?; + + run_blocking(move || match receiver.recv() { + Ok(Ok(completion)) => Ok(completion), + Ok(Err(e)) => Err(e), + Err(e) => Err(format!("endpoint worker error: {e}")), + }) + .await + } + + async fn clear_halt(&self) -> Result<()> { + let (response, receiver) = mpsc::channel(); + self.sender + .send(EndpointCommand::ClearHalt { response }) + .map_err(|_| napi::Error::from_reason("endpoint worker stopped"))?; + + run_blocking(move || match receiver.recv() { + Ok(result) => result, + Err(e) => Err(format!("endpoint worker error: {e}")), + }) + .await + } + + fn stop(&self) { + let _ = self.sender.send(EndpointCommand::Stop); + if let Some(join_handle) = self + .join_handle + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + { + let _ = join_handle.join(); + } + } +} + +fn handle_endpoint_command( + command: EndpointCommand, + endpoint: &mut AnyEndpoint, + pending: &mut VecDeque, + clear_halt_responses: &mut Vec, + stopping: &mut bool, +) { + match command { + EndpointCommand::Transfer { + buffer, + timeout, + response, + } => { + if *stopping { + let _ = response.send(Err("endpoint worker stopped".to_string())); + return; + } + + endpoint.submit(buffer); + pending.push_back(PendingTransfer { + deadline: Instant::now() + timeout, + response: Some(response), + }); + } + EndpointCommand::ClearHalt { response } => { + endpoint.cancel_all(); + cancel_pending_transfers(pending); + clear_halt_responses.push(response); + } + EndpointCommand::Stop => { + *stopping = true; + endpoint.cancel_all(); + cancel_pending_transfers(pending); + for response in clear_halt_responses.drain(..) { + let _ = response.send(Err("endpoint worker stopped".to_string())); + } + } + } +} + +fn cancel_pending_transfers(pending: &mut VecDeque) { + for pending_transfer in pending { + if let Some(response) = pending_transfer.response.take() { + let _ = response.send(Err(format!( + "{:?}", + nusb::transfer::TransferError::Cancelled + ))); + } + } +} + +fn expire_timed_out_transfers(pending: &mut VecDeque) { + let now = Instant::now(); + for pending_transfer in pending { + if pending_transfer.deadline <= now { + if let Some(response) = pending_transfer.response.take() { + let _ = response.send(Err(format!( + "{:?}", + nusb::transfer::TransferError::Cancelled + ))); + } + } + } +} + +fn next_wait(pending: &VecDeque) -> Duration { + const ENDPOINT_POLL_TIMEOUT: Duration = Duration::from_millis(10); + + pending + .iter() + .filter(|pending_transfer| pending_transfer.response.is_some()) + .map(|pending_transfer| { + pending_transfer + .deadline + .saturating_duration_since(Instant::now()) + }) + .min() + .map(|timeout| timeout.min(ENDPOINT_POLL_TIMEOUT)) + .unwrap_or(ENDPOINT_POLL_TIMEOUT) +} + #[napi(object)] pub struct UsbEndpoint { #[napi(writable = false)] @@ -310,8 +582,8 @@ pub struct UsbDevice { device_info: nusb::DeviceInfo, device: Option, interfaces: Vec>, - in_endpoints: SharedEndpointCache>, - out_endpoints: SharedEndpointCache>, + in_endpoints: SharedEndpointCache>, + out_endpoints: SharedEndpointCache>, #[napi(writable = false)] pub vendorId: u16, @@ -535,7 +807,6 @@ impl UsbDevice { pub async fn selectConfiguration(&self, configurationValue: u8) -> Result<()> { match &self.device { Some(device) => { - self.clear_endpoint_caches(); let found = device .configurations() .any(|c| c.configuration_value() == configurationValue); @@ -552,6 +823,7 @@ impl UsbDevice { } #[cfg(not(windows))] { + self.clear_endpoint_caches(); let device = device.clone(); run_blocking(move || { device @@ -583,7 +855,6 @@ impl UsbDevice { }) .await?; self.interfaces[interfaceNumber as usize] = Some(interface); - self.clear_endpoint_caches(); Ok(()) } None => Err(napi::Error::from_reason( @@ -727,22 +998,17 @@ impl UsbDevice { length: u32, ) -> Result> { match self.get_in_endpoint(endpointNumber) { - Some(endpoint) => { - let v = run_blocking(move || { - let mut endpoint = endpoint_guard(&endpoint); - let packet_size = endpoint.max_packet_size(); - let req = (((length as usize) + packet_size - 1) / packet_size) * packet_size; - let buf = Buffer::new(req); - let completion = - endpoint.transfer_blocking(buf, Duration::from_millis(timeout as u64)); - completion - .status - .map_err(|e| format!("transferIn error: {e:?}"))?; - let mut v = completion.buffer.into_vec(); - v.truncate(completion.actual_len.min(length as usize)); - Ok(v) - }) - .await?; + Some(worker) => { + let packet_size = worker.max_packet_size; + let req = (((length as usize) + packet_size - 1) / packet_size) * packet_size; + let completion = worker + .transfer(Buffer::new(req), Duration::from_millis(timeout as u64)) + .await?; + completion + .status + .map_err(|e| napi::Error::from_reason(format!("transferIn error: {e:?}")))?; + let mut v = completion.buffer.into_vec(); + v.truncate(completion.actual_len.min(length as usize)); Ok(Some(Uint8Array::from(v))) } None => { @@ -761,20 +1027,17 @@ impl UsbDevice { data: Uint8Array, ) -> Result { match self.get_out_endpoint(endpointNumber) { - Some(endpoint) => { + Some(worker) => { let data = data.to_vec(); - run_blocking(move || { - let mut endpoint = endpoint_guard(&endpoint); - let mut buf = Buffer::new(data.len()); - buf.extend_from_slice(&data); - let completion = - endpoint.transfer_blocking(buf, Duration::from_millis(timeout as u64)); - completion - .status - .map_err(|e| format!("transferOut error: {e:?}"))?; - Ok(completion.actual_len as u32) - }) - .await + let mut buf = Buffer::new(data.len()); + buf.extend_from_slice(&data); + let completion = worker + .transfer(buf, Duration::from_millis(timeout as u64)) + .await?; + completion + .status + .map_err(|e| napi::Error::from_reason(format!("transferOut error: {e:?}")))?; + Ok(completion.actual_len as u32) } None => { return Err(napi::Error::from_reason( @@ -823,14 +1086,11 @@ impl UsbDevice { ) -> Result<()> { if direction == "in" { match self.get_in_endpoint(endpointNumber) { - Some(endpoint) => { - run_blocking(move || { - let mut endpoint = endpoint_guard(&endpoint); - endpoint - .clear_halt_blocking() - .map_err(|e| format!("clearHalt error: {e}")) - }) - .await?; + Some(worker) => { + worker + .clear_halt() + .await + .map_err(|e| napi::Error::from_reason(format!("clearHalt error: {e}")))?; } None => { return Err(napi::Error::from_reason( @@ -840,14 +1100,11 @@ impl UsbDevice { } } else { match self.get_out_endpoint(endpointNumber) { - Some(endpoint) => { - run_blocking(move || { - let mut endpoint = endpoint_guard(&endpoint); - endpoint - .clear_halt_blocking() - .map_err(|e| format!("clearHalt error: {e}")) - }) - .await?; + Some(worker) => { + worker + .clear_halt() + .await + .map_err(|e| napi::Error::from_reason(format!("clearHalt error: {e}")))?; } None => { return Err(napi::Error::from_reason( @@ -944,18 +1201,12 @@ impl UsbDevice { None } - fn get_in_endpoint( - &self, - endpointNumber: u8, - ) -> Option>>> { + fn get_in_endpoint(&self, endpointNumber: u8) -> Option> { self.in_endpoints .get_or_try_insert_with(endpointNumber, || self.open_endpoint(endpointNumber)) } - fn get_out_endpoint( - &self, - endpointNumber: u8, - ) -> Option>>> { + fn get_out_endpoint(&self, endpointNumber: u8) -> Option> { self.out_endpoints .get_or_try_insert_with(endpointNumber, || self.open_endpoint(endpointNumber)) }