Skip to content

Use win32 threadpool for IO events, support control transfer timeouts #171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
kevinmehall wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

Use win32 threadpool for IO events, support control transfer timeouts #171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3e24e0e
windows: Use Win32 threadpool for IO callbacks instead of dedicated t…
kevinmehall Nov 3, 2025
3984208
Use NonNull instead of Box for Idle transfer
kevinmehall Nov 3, 2025
ed0b7b3
windows: Implement control transfer timeouts
kevinmehall Nov 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ rustix = { version = "1.0.1", features = ["fs", "event", "net", "time", "mm"] }
linux-raw-sys = { version = ">= 0.11, <= 0.12", features = ["ioctl"] }

[target.'cfg(target_os="windows")'.dependencies]
windows-sys = { version = ">= 0.60, <= 0.61", features = ["Win32_Devices_Usb", "Win32_Devices_DeviceAndDriverInstallation", "Win32_Foundation", "Win32_Devices_Properties", "Win32_Storage_FileSystem", "Win32_Security", "Win32_System_IO", "Win32_System_Registry", "Win32_System_Com"] }
windows-sys = { version = ">= 0.60, <= 0.61", features = ["Win32_Devices_Usb", "Win32_Devices_DeviceAndDriverInstallation", "Win32_Foundation", "Win32_Devices_Properties", "Win32_Storage_FileSystem", "Win32_Security", "Win32_System_IO", "Win32_System_Registry", "Win32_System_Threading", "Win32_System_Com"] }

[target.'cfg(target_os="macos")'.dependencies]
core-foundation = "0.10.1"
Expand Down
4 changes: 0 additions & 4 deletions src/device.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,8 +426,6 @@ impl Interface {
/// byte of `index` must match the interface number, or
/// `TransferError::InvalidArgument` will be returned. This is a WinUSB
/// limitation.
/// * On Windows, the timeout is currently fixed to 5 seconds and the
/// timeout argument is ignored.
pub fn control_in(
&self,
data: ControlIn,
Expand Down Expand Up @@ -467,8 +465,6 @@ impl Interface {
/// byte of `index` must match the interface number, or
/// `TransferError::InvalidArgument` will be returned. This is a WinUSB
/// limitation.
/// * On Windows, the timeout is currently fixed to 5 seconds and the
/// timeout argument is ignored.
pub fn control_out(
&self,
data: ControlOut,
Expand Down
170 changes: 154 additions & 16 deletions src/platform/windows_winusb/device.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,8 @@ use std::{
collections::{btree_map::Entry, BTreeMap, VecDeque},
ffi::c_void,
io,
mem::{size_of_val, transmute},
os::windows::{
io::{AsRawHandle, RawHandle},
prelude::OwnedHandle,
},
mem::{self, size_of_val, transmute},
os::windows::io::{AsRawHandle, OwnedHandle, RawHandle},
ptr::{self, null_mut},
sync::{Arc, Mutex},
task::{Context, Poll},
Expand All @@ -24,9 +21,16 @@ use windows_sys::Win32::{
Foundation::{
GetLastError, ERROR_BAD_COMMAND, ERROR_DEVICE_NOT_CONNECTED, ERROR_FILE_NOT_FOUND,
ERROR_IO_PENDING, ERROR_NOT_FOUND, ERROR_NO_MORE_ITEMS, ERROR_NO_SUCH_DEVICE, FALSE,
HANDLE, TRUE,
FILETIME, HANDLE, TRUE,
},
System::{
Threading::{
CancelThreadpoolIo, CloseThreadpoolIo, CreateThreadpoolIo, CreateThreadpoolTimer,
SetThreadpoolTimer, StartThreadpoolIo, WaitForThreadpoolTimerCallbacks,
PTP_CALLBACK_INSTANCE, PTP_IO, PTP_TIMER,
},
IO::{CancelIoEx, OVERLAPPED},
},
System::IO::{CancelIoEx, OVERLAPPED},
};

use crate::{
Expand Down Expand Up @@ -263,6 +267,7 @@ impl BitSet256 {
pub(crate) struct WinusbFileHandle {
first_interface: u8,
handle: OwnedHandle,
threadpool_io: PTP_IO,
winusb_handle: WINUSB_INTERFACE_HANDLE,
claimed_interfaces: BitSet256,
}
Expand All @@ -275,11 +280,27 @@ impl WinusbFileHandle {
fn new(path: &WCStr, first_interface: u8) -> Result<Self, Error> {
let handle = create_file(path)
.map_err(|e| Error::new_os(ErrorKind::Other, "failed to open device", e).log_debug())?;
super::events::register(&handle)?;

let threadpool_io = unsafe {
CreateThreadpoolIo(
raw_handle(&handle),
Some(io_callback),
null_mut(),
null_mut(),
)
};

if threadpool_io == 0 {
let err = unsafe { GetLastError() };
return Err(
Error::new_os(ErrorKind::Other, "CreateThreadpoolIo failed", err).log_error(),
);
}

let winusb_handle = unsafe {
let mut h = ptr::null_mut();
if WinUsb_Initialize(raw_handle(&handle), &mut h) == FALSE {
CloseThreadpoolIo(threadpool_io);
return Err(Error::new_os(
ErrorKind::Other,
"failed to initialize WinUSB",
Expand All @@ -292,9 +313,27 @@ impl WinusbFileHandle {

debug!("Opened WinUSB handle for {path} (interface {first_interface})");

unsafe {
// Disable WinUSB's default control transfer timeout so we can do our own
// per-request timeout handling by cancelling the transfer with a timer.
let timeout: u32 = 0;
let r = WinUsb_SetPipePolicy(
winusb_handle,
0x00,
Usb::PIPE_TRANSFER_TIMEOUT,
size_of_val(&timeout) as u32,
&timeout as *const _ as *const c_void,
);
if r != TRUE {
let err = GetLastError();
warn!("Failed to disable default timeout on control endpoint, error {err}");
}
}

Ok(WinusbFileHandle {
first_interface,
handle,
threadpool_io,
winusb_handle,
claimed_interfaces: BitSet256::new(),
})
Expand Down Expand Up @@ -349,6 +388,7 @@ impl WinusbFileHandle {

Ok(Arc::new(WindowsInterface {
handle: self.handle.as_raw_handle(),
threadpool_io: self.threadpool_io,
device: device.clone(),
interface_number,
first_interface_number: self.first_interface,
Expand All @@ -365,16 +405,78 @@ impl Drop for WinusbFileHandle {
self.first_interface
);
unsafe {
CloseThreadpoolIo(self.threadpool_io);
WinUsb_Free(self.winusb_handle);
}
}
}

unsafe extern "system" fn io_callback(
_instance: PTP_CALLBACK_INSTANCE,
_context: *mut c_void,
overlapped: *mut c_void,
result: u32,
bytes_transferred: usize,
_io: PTP_IO,
) {
let t = overlapped as *mut TransferData;
{
let transfer = unsafe { &*t };

debug!(
"Transfer {t:?} on endpoint {:02x} complete: status {}, {} bytes",
transfer.endpoint, result, bytes_transferred,
);

if let Some(timer) = transfer.timeout {
// Cancel the timeout and wait for any callback to complete that may be concurrently
// accessing `transfer`.
unsafe {
SetThreadpoolTimer(timer, ptr::null_mut(), 0, 0);
WaitForThreadpoolTimerCallbacks(timer, 1);
}
}
}
unsafe { notify_completion::<TransferData>(t) }
}

unsafe extern "system" fn timer_callback(
_instance: PTP_CALLBACK_INSTANCE,
context: *mut c_void,
_timer: PTP_TIMER,
) {
let transfer_data = &*(context as *const TransferData);
debug!(
"Transfer {context:?} timeout on endpoint 0x{:02X}",
transfer_data.endpoint
);
unsafe {
CancelIoEx(transfer_data.handle, &transfer_data.overlapped);
}
}

fn duration_to_filetime(duration: Duration) -> FILETIME {
let time = i64::try_from(duration.as_micros())
.unwrap_or(i64::MAX)
.saturating_mul(-10); // in 100-nanosecond intervals, negative for relative time
FILETIME {
dwLowDateTime: (time & 0xFFFFFFFF) as u32,
dwHighDateTime: (time >> 32) as u32,
}
}

pub(crate) struct WindowsInterface {
/// Owned by the `WinUSBFileHandle`
pub(crate) handle: RawHandle,

/// Owned by the `WinUSBFileHandle`
pub(crate) threadpool_io: PTP_IO,

pub(crate) device: Arc<WindowsDevice>,
pub(crate) first_interface_number: u8,
pub(crate) interface_number: u8,

/// Owned by this object if `first_interface_number != interface_number`, otherwise owned by the `WinUSBFileHandle`
pub(crate) winusb_handle: WINUSB_INTERFACE_HANDLE,
state: Mutex<InterfaceState>,
}
Expand Down Expand Up @@ -430,7 +532,7 @@ impl WindowsInterface {
data: ControlIn,
timeout: Duration,
) -> impl MaybeFuture<Output = Result<Vec<u8>, TransferError>> {
let mut t = TransferData::new(0x80);
let mut t = TransferData::new(self.handle, 0x80);
t.set_buffer(Buffer::new(data.length as usize));

let pkt = WINUSB_SETUP_PACKET {
Expand All @@ -443,7 +545,7 @@ impl WindowsInterface {

let intf = self.clone();

TransferFuture::new(t, |t| self.submit_control(t, pkt)).map(move |mut t| {
TransferFuture::new(t, |t| self.submit_control(t, pkt, timeout)).map(move |mut t| {
let c = t.take_completion(&intf);
c.status?;
Ok(c.buffer.into_vec())
Expand All @@ -455,7 +557,7 @@ impl WindowsInterface {
data: ControlOut,
timeout: Duration,
) -> impl MaybeFuture<Output = Result<(), TransferError>> {
let mut t = TransferData::new(0x00);
let mut t = TransferData::new(self.handle, 0x00);
t.set_buffer(Buffer::from(data.data.to_vec()));

let pkt = WINUSB_SETUP_PACKET {
Expand All @@ -468,7 +570,7 @@ impl WindowsInterface {

let intf = self.clone();

TransferFuture::new(t, |t| self.submit_control(t, pkt)).map(move |mut t| {
TransferFuture::new(t, |t| self.submit_control(t, pkt, timeout)).map(move |mut t| {
let c = t.take_completion(&intf);
c.status
})
Expand Down Expand Up @@ -560,14 +662,18 @@ impl WindowsInterface {
let dir = Direction::from_address(endpoint);
let len = t.request_len;
let buf = t.buf;
t.overlapped.InternalHigh = 0;
t.overlapped = unsafe { mem::zeroed() };
t.error_from_submit = Ok(());

let t = t.pre_submit();
let ptr = t.as_ptr();

debug!("Submit transfer {ptr:?} on endpoint {endpoint:02X} for {len} bytes {dir:?}");

unsafe {
StartThreadpoolIo(self.threadpool_io);
}

let r = unsafe {
match dir {
Direction::Out => WinUsb_WritePipe(
Expand Down Expand Up @@ -596,12 +702,13 @@ impl WindowsInterface {
&self,
mut t: Idle<TransferData>,
pkt: WINUSB_SETUP_PACKET,
timeout: Duration,
) -> Pending<TransferData> {
let endpoint = t.endpoint;
let dir = Direction::from_address(endpoint);
let len = t.request_len;
let buf = t.buf;
t.overlapped.InternalHigh = 0;
t.overlapped = unsafe { mem::zeroed() };
t.error_from_submit = Ok(());

if pkt.RequestType & 0x1f == Recipient::Interface as u8
Expand All @@ -612,11 +719,35 @@ impl WindowsInterface {
return t.simulate_complete();
}

let t = t.pre_submit();
let ptr = t.as_ptr();
let timer = unsafe {
CreateThreadpoolTimer(
Some(timer_callback),
ptr as *mut core::ffi::c_void,
ptr::null_mut(),
)
};
if timer == 0 {
let e = unsafe { GetLastError() };
log::error!("CreateThreadpoolTimer failed with error {e}");
} else {
t.timeout = Some(timer);
}

let t = t.pre_submit();

// Start the timeout, now that we're no longer borrowing the transfer, but also before submitting the transfer:
// this guarantees that the IO callback trying to clear the timer can't occur before `SetThreadpoolTimer`
// enables it.
let tm = duration_to_filetime(timeout);
unsafe { SetThreadpoolTimer(timer, &tm, 0, 0) };

debug!("Submit control {dir:?} transfer {ptr:?} for {len} bytes");

unsafe {
StartThreadpoolIo(self.threadpool_io);
}

let r = unsafe {
WinUsb_ControlTransfer(
self.winusb_handle,
Expand All @@ -641,6 +772,10 @@ impl WindowsInterface {
if err != ERROR_IO_PENDING {
error!("submit failed: {}", io::Error::from_raw_os_error(err as _));

unsafe {
CancelThreadpoolIo(self.threadpool_io);
}

// Safety: Transfer was not submitted, so we still own it
// and must complete it in place of the event thread.
unsafe {
Expand Down Expand Up @@ -711,7 +846,10 @@ impl WindowsEndpoint {

fn make_transfer(&mut self, buffer: Buffer) -> Idle<TransferData> {
let mut t = self.idle_transfer.take().unwrap_or_else(|| {
Idle::new(self.inner.clone(), TransferData::new(self.inner.address))
Idle::new(
self.inner.clone(),
TransferData::new(self.inner.interface.handle, self.inner.address),
)
});
t.set_buffer(buffer);
t
Expand Down
Loading
Loading