Error on IN transfer that is not a multiple of max packet size on all platforms

src/device.rs

@@ -10,7 +10,7 @@ use crate::{
     },
     DeviceInfo, Error, MaybeFuture, Speed,
 };
-use log::error;
+use log::{error, warn};
 use std::{
     future::{poll_fn, Future},
     io::ErrorKind,
@@ -635,10 +635,26 @@ impl<EpType: BulkOrInterrupt, Dir: EndpointDirection> Endpoint<EpType, Dir> {
     /// For an OUT transfer, the buffer's `len` is the number of bytes
     /// initialized, which will be sent to the device.
     ///
-    /// For an IN transfer, the buffer's `requested_len` is the number of
-    /// bytes requested. It must be a multiple of the endpoint's [maximum packet
-    /// size][`Self::max_packet_size`].
+    /// For an IN transfer, the buffer's `requested_len` is the number of bytes
+    /// requested. It must be a nonzero multiple of the endpoint's [maximum
+    /// packet size][`Self::max_packet_size`] or the transfer will fail with
+    /// `TransferError::InvalidArgument`. Up to `requested_len /
+    /// max_packet_size` packets will be received, ending early when any packet
+    /// is shorter than `max_packet_size`.
     pub fn submit(&mut self, buf: Buffer) {
+        if Dir::DIR == Direction::In {
+            let req_len = buf.requested_len();
+            if req_len == 0 || req_len % self.max_packet_size() != 0 {
+                warn!(
+                    "Submitting transfer with length {req_len} which is not a multiple of max packet size {} on IN endpoint {:02x}",
+                    self.max_packet_size(),
+                    self.endpoint_address(),
+                );
+
+                return self.backend.submit_err(buf, TransferError::InvalidArgument);
+            }
+        }
+
         self.backend.submit(buf)
     }
 
@@ -647,6 +663,12 @@ impl<EpType: BulkOrInterrupt, Dir: EndpointDirection> Endpoint<EpType, Dir> {
     /// This future is cancel-safe: it can be cancelled and re-created without
     /// side effects, enabling its use in `select!{}` or similar.
     ///
+    /// An OUT transfer completes when the specified data has been sent or an
+    /// error occurs. An IN transfer completes when a packet smaller than
+    /// `max_packet_size` is received, the full `requested_len` is received
+    /// (without waiting for or consuming any subsequent zero-length packet), or
+    /// an error occurs.
+    ///
     /// ## Panics
     /// * if there are no transfers pending (that is, if [`Self::pending()`]
     ///   would return 0).

src/platform/linux_usbfs/device.rs

@@ -691,18 +691,30 @@ impl LinuxEndpoint {
         }
     }
 
-    pub(crate) fn submit(&mut self, data: Buffer) {
-        let mut transfer = self.idle_transfer.take().unwrap_or_else(|| {
+    fn get_transfer(&mut self) -> Idle<TransferData> {
+        self.idle_transfer.take().unwrap_or_else(|| {
             Idle::new(
                 self.inner.clone(),
                 super::TransferData::new(self.inner.address, self.inner.ep_type),
             )
-        });
+        })
+    }
+
+    pub(crate) fn submit(&mut self, data: Buffer) {
+        let mut transfer = self.get_transfer();
         transfer.set_buffer(data);
         self.pending
             .push_back(self.inner.interface.device.submit(transfer));
     }
 
+    pub(crate) fn submit_err(&mut self, data: Buffer, error: TransferError) {
+        assert_eq!(error, TransferError::InvalidArgument);
+        let mut transfer = self.get_transfer();
+        transfer.set_buffer(data);
+        transfer.urb_mut().status = Errno::INVAL.raw_os_error();
+        self.pending.push_back(transfer.simulate_complete());
+    }
+
     pub(crate) fn poll_next_complete(&mut self, cx: &mut Context) -> Poll<Completion> {
         self.inner.notify.subscribe(cx);
         if let Some(mut transfer) = take_completed_from_queue(&mut self.pending) {

src/platform/linux_usbfs/mod.rs

@@ -31,6 +31,7 @@ fn errno_to_transfer_error(e: Errno) -> TransferError {
         Errno::PROTO | Errno::ILSEQ | Errno::OVERFLOW | Errno::COMM | Errno::TIME => {
             TransferError::Fault
         }
+        Errno::INVAL => TransferError::InvalidArgument,
         _ => TransferError::Unknown(e.raw_os_error() as u32),
     }
 }

src/platform/macos_iokit/device.rs

@@ -466,25 +466,30 @@ impl MacEndpoint {
         );
     }
 
-    pub(crate) fn submit(&mut self, buffer: Buffer) {
+    fn make_transfer(&mut self, buffer: Buffer) -> Idle<TransferData> {
         let mut transfer = self
             .idle_transfer
             .take()
             .unwrap_or_else(|| Idle::new(self.inner.clone(), super::TransferData::new()));
 
         let buffer = ManuallyDrop::new(buffer);
-        let endpoint = self.inner.address;
-        let dir = Direction::from_address(endpoint);
-        let pipe_ref = self.inner.pipe_ref;
-
         transfer.buf = buffer.ptr;
         transfer.capacity = buffer.capacity;
         transfer.actual_len = 0;
-        let req_len = match dir {
+        let req_len = match Direction::from_address(self.inner.address) {
             Direction::Out => buffer.len,
             Direction::In => buffer.requested_len,
         };
         transfer.requested_len = req_len;
+        transfer
+    }
+
+    pub(crate) fn submit(&mut self, buffer: Buffer) {
+        let transfer = self.make_transfer(buffer);
+        let endpoint = self.inner.address;
+        let dir = Direction::from_address(endpoint);
+        let req_len = transfer.requested_len;
+        let buf_ptr = transfer.buf;
 
         let transfer = transfer.pre_submit();
         let ptr = transfer.as_ptr();
@@ -494,9 +499,9 @@ impl MacEndpoint {
                 Direction::Out => call_iokit_function!(
                     self.inner.interface.interface.raw,
                     WritePipeAsync(
-                        pipe_ref,
-                        buffer.ptr as *mut c_void,
-                        buffer.len,
+                        self.inner.pipe_ref,
+                        buf_ptr as *mut c_void,
+                        req_len,
                         transfer_callback,
                         ptr as *mut c_void
                     )
@@ -504,9 +509,9 @@ impl MacEndpoint {
                 Direction::In => call_iokit_function!(
                     self.inner.interface.interface.raw,
                     ReadPipeAsync(
-                        pipe_ref,
-                        buffer.ptr as *mut c_void,
-                        buffer.requested_len,
+                        self.inner.pipe_ref,
+                        buf_ptr as *mut c_void,
+                        req_len,
                         transfer_callback,
                         ptr as *mut c_void
                     )
@@ -519,7 +524,7 @@ impl MacEndpoint {
                 "Submitted {dir:?} transfer {ptr:?} of len {req_len} on endpoint {endpoint:02X}"
             );
         } else {
-            error!("Failed to submit transfer {ptr:?} of len {req_len} on endpoint {endpoint:02X}: {res:x}");
+            error!("Failed to submit {dir:?} transfer {ptr:?} of len {req_len} on endpoint {endpoint:02X}: {res:x}");
             unsafe {
                 // Complete the transfer in the place of the callback
                 (*ptr).status = res;
@@ -530,6 +535,13 @@ impl MacEndpoint {
         self.pending.push_back(transfer);
     }
 
+    pub(crate) fn submit_err(&mut self, buffer: Buffer, err: TransferError) {
+        assert_eq!(err, TransferError::InvalidArgument);
+        let mut transfer = self.make_transfer(buffer);
+        transfer.status = io_kit_sys::ret::kIOReturnBadArgument;
+        self.pending.push_back(transfer.simulate_complete());
+    }
+
     pub(crate) fn poll_next_complete(&mut self, cx: &mut Context) -> Poll<Completion> {
         self.inner.notify.subscribe(cx);
         if let Some(mut transfer) = take_completed_from_queue(&mut self.pending) {

src/platform/macos_iokit/mod.rs

@@ -33,6 +33,7 @@ fn status_to_transfer_result(status: IOReturn) -> Result<(), TransferError> {
             Err(TransferError::Cancelled)
         }
         iokit_c::kIOUSBPipeStalled => Err(TransferError::Stall),
+        io_kit_sys::ret::kIOReturnBadArgument => Err(TransferError::InvalidArgument), // used for `submit_err`
         _ => Err(TransferError::Unknown(status as u32)),
     }
 }

src/platform/windows_winusb/device.rs

@@ -578,23 +578,17 @@ impl WindowsInterface {
         t.overlapped.InternalHigh = 0;
         t.error_from_submit = Ok(());
 
-        let t = t.pre_submit();
-        let ptr = t.as_ptr();
-
         if pkt.RequestType & 0x1f == Recipient::Interface as u8
             && pkt.Index as u8 != self.interface_number
         {
             warn!("WinUSB requires control transfer with `Recipient::Interface` to pass the interface number in `index`");
-
-            // Safety: Transfer is not submitted, so we can complete it in place of the event thread.
-            unsafe {
-                (*t.as_ptr()).error_from_submit = Err(TransferError::InvalidArgument);
-                notify_completion::<TransferData>(t.as_ptr());
-            }
-
-            return t;
+            t.error_from_submit = Err(TransferError::InvalidArgument);
+            return t.simulate_complete();
         }
 
+        let t = t.pre_submit();
+        let ptr = t.as_ptr();
+
         debug!("Submit control {dir:?} transfer {ptr:?} for {len} bytes");
 
         let r = unsafe {
@@ -689,15 +683,26 @@ impl WindowsEndpoint {
         }
     }
 
-    pub(crate) fn submit(&mut self, buffer: Buffer) {
+    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))
         });
         t.set_buffer(buffer);
+        t
+    }
+
+    pub(crate) fn submit(&mut self, buffer: Buffer) {
+        let t = self.make_transfer(buffer);
         let t = self.inner.interface.submit(t);
         self.pending.push_back(t);
     }
 
+    pub(crate) fn submit_err(&mut self, buffer: Buffer, err: TransferError) {
+        let mut t = self.make_transfer(buffer);
+        t.error_from_submit = Err(err);
+        self.pending.push_back(t.simulate_complete());
+    }
+
     pub(crate) fn poll_next_complete(&mut self, cx: &mut Context) -> Poll<Completion> {
         self.inner.notify.subscribe(cx);
         if let Some(mut transfer) = take_completed_from_queue(&mut self.pending) {

src/transfer/internal.rs

@@ -128,6 +128,12 @@ impl<P> Idle<P> {
             ptr: unsafe { NonNull::new_unchecked(Box::into_raw(self.0)) },
         }
     }
+
+    pub(crate) fn simulate_complete(self) -> Pending<P> {
+        Pending {
+            ptr: unsafe { NonNull::new_unchecked(Box::into_raw(self.0)) },
+        }
+    }
 }
 
 impl<P> Deref for Idle<P> {