Support USB transfers from arbitrary threads in the new Linux backend.

device/usb/usb_device_handle_usbfs.cc

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "device/usb/usb_device_handle_usbfs.h"
 
 #if defined(OS_ANDROID) && __ANDROID_API__ < 21
 #include <linux/usb_ch9.h>
 #else
 #include <linux/usb/ch9.h>
@@ skipping 211 matching lines @@
 }
 
 void UsbDeviceHandleUsbfs::FileThreadHelper::WillDestroyCurrentMessageLoop() {
   DCHECK(thread_checker_.CalledOnValidThread());
   delete this;
 }
 
 struct UsbDeviceHandleUsbfs::Transfer {
   Transfer() = delete;
   Transfer(scoped_refptr<net::IOBuffer> buffer,
-           const TransferCallback& callback);
+           const TransferCallback& callback,
+           scoped_refptr<base::SingleThreadTaskRunner> callback_runner);
   Transfer(scoped_refptr<net::IOBuffer> buffer,
            const IsochronousTransferCallback& callback);
   ~Transfer();
 
   void* operator new(std::size_t size, size_t number_of_iso_packets);
+  void RunCallback(UsbTransferStatus status, size_t bytes_transferred);
+  void RunIsochronousCallback(const std::vector<IsochronousPacket>& packets);
 
   scoped_refptr<net::IOBuffer> control_transfer_buffer;
   scoped_refptr<net::IOBuffer> buffer;
   TransferCallback callback;
   IsochronousTransferCallback isoc_callback;
+  scoped_refptr<base::SingleThreadTaskRunner> callback_runner;
   base::CancelableClosure timeout_closure;
   bool cancelled = false;
 
   // The |urb| field must be the last in the struct so that the extra space
   // allocated by the overridden new function above extends the length of its
   // |iso_frame_desc| field.
   usbdevfs_urb urb;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(Transfer);
 };
 
-UsbDeviceHandleUsbfs::Transfer::Transfer(scoped_refptr<net::IOBuffer> buffer,
-                                         const TransferCallback& callback)
-    : buffer(buffer), callback(callback) {
+UsbDeviceHandleUsbfs::Transfer::Transfer(
+    scoped_refptr<net::IOBuffer> buffer,
+    const TransferCallback& callback,
+    scoped_refptr<base::SingleThreadTaskRunner> callback_runner)
+    : buffer(buffer), callback(callback), callback_runner(callback_runner) {
   memset(&urb, 0, sizeof(urb));
   urb.usercontext = this;
   urb.buffer = buffer->data();
 }
 
 UsbDeviceHandleUsbfs::Transfer::Transfer(
     scoped_refptr<net::IOBuffer> buffer,
     const IsochronousTransferCallback& callback)
     : buffer(buffer), isoc_callback(callback) {
   memset(&urb, 0, sizeof(urb) +
                       sizeof(usbdevfs_iso_packet_desc) * urb.number_of_packets);
   urb.usercontext = this;
   urb.buffer = buffer->data();
 }
 
 UsbDeviceHandleUsbfs::Transfer::~Transfer() = default;
 
 void* UsbDeviceHandleUsbfs::Transfer::operator new(
     std::size_t size,
     size_t number_of_iso_packets) {
   void* p = ::operator new(
       size + sizeof(usbdevfs_iso_packet_desc) * number_of_iso_packets);
   Transfer* transfer = static_cast<Transfer*>(p);
   transfer->urb.number_of_packets = number_of_iso_packets;
   return p;
 }
 
+void UsbDeviceHandleUsbfs::Transfer::RunCallback(UsbTransferStatus status,
+                                                 size_t bytes_transferred) {
+  DCHECK_NE(urb.type, USBDEVFS_URB_TYPE_ISO);
+  DCHECK(!callback.is_null());
+  if (!callback_runner || callback_runner->BelongsToCurrentThread()) {
+    callback.Run(status, buffer, bytes_transferred);
+  } else {
+    callback_runner->PostTask(
+        FROM_HERE, base::Bind(callback, status, buffer, bytes_transferred));
+  }
+}
+
+void UsbDeviceHandleUsbfs::Transfer::RunIsochronousCallback(
+    const std::vector<IsochronousPacket>& packets) {
+  DCHECK_EQ(urb.type, USBDEVFS_URB_TYPE_ISO);
+  DCHECK(!isoc_callback.is_null());
+  isoc_callback.Run(buffer, packets);
+}
+
 UsbDeviceHandleUsbfs::UsbDeviceHandleUsbfs(
     scoped_refptr<UsbDevice> device,
     base::ScopedFD fd,
     scoped_refptr<base::SequencedTaskRunner> blocking_task_runner)
     : device_(device),
       fd_(std::move(fd)),
       blocking_task_runner_(blocking_task_runner) {
   DCHECK(device_);
   DCHECK(fd_.is_valid());
   DCHECK(blocking_task_runner_);
@@ skipping 113 matching lines @@
                                            scoped_refptr<net::IOBuffer> buffer,
                                            size_t length,
                                            unsigned int timeout,
                                            const TransferCallback& callback) {
   if (!device_) {
     task_runner_->PostTask(
         FROM_HERE, base::Bind(callback, USB_TRANSFER_DISCONNECT, nullptr, 0));
     return;
   }
 
-  std::unique_ptr<Transfer> transfer(new (0) Transfer(buffer, callback));
+  std::unique_ptr<Transfer> transfer(new (0)
+                                         Transfer(buffer, callback, nullptr));
   transfer->control_transfer_buffer =
       BuildControlTransferBuffer(direction, request_type, recipient, request,
                                  value, index, buffer, length);
   transfer->urb.type = USBDEVFS_URB_TYPE_CONTROL;
   transfer->urb.endpoint = 0;
   transfer->urb.buffer = transfer->control_transfer_buffer->data();
   transfer->urb.buffer_length = 8 + length;
 
   // USBDEVFS_SUBMITURB appears to be non-blocking as completion is reported
   // by USBDEVFS_REAPURBNDELAY.
@@ skipping 10 matching lines @@
 }
 
 void UsbDeviceHandleUsbfs::IsochronousTransferIn(
     uint8_t endpoint_number,
     const std::vector<uint32_t>& packet_lengths,
     unsigned int timeout,
     const IsochronousTransferCallback& callback) {
   uint8_t endpoint_address = USB_DIR_IN | endpoint_number;
   size_t total_length =
       std::accumulate(packet_lengths.begin(), packet_lengths.end(), 0u);
-  IsochronousTransferInternal(endpoint_address, new net::IOBuffer(total_length),
-                              total_length, packet_lengths, timeout, callback);
+  scoped_refptr<net::IOBuffer> buffer(new net::IOBuffer(total_length));
+  IsochronousTransferInternal(endpoint_address, buffer, total_length,
+                              packet_lengths, timeout, callback);
 }
 
 void UsbDeviceHandleUsbfs::IsochronousTransferOut(
     uint8_t endpoint_number,
     scoped_refptr<net::IOBuffer> buffer,
     const std::vector<uint32_t>& packet_lengths,
     unsigned int timeout,
     const IsochronousTransferCallback& callback) {
   uint8_t endpoint_address = USB_DIR_OUT | endpoint_number;
   size_t total_length =
       std::accumulate(packet_lengths.begin(), packet_lengths.end(), 0u);
   IsochronousTransferInternal(endpoint_address, buffer, total_length,
                               packet_lengths, timeout, callback);
 }
 
 void UsbDeviceHandleUsbfs::GenericTransfer(UsbEndpointDirection direction,
                                            uint8_t endpoint_number,
                                            scoped_refptr<net::IOBuffer> buffer,
                                            size_t length,
                                            unsigned int timeout,
                                            const TransferCallback& callback) {
-  if (!device_) {
+  if (task_runner_->BelongsToCurrentThread()) {
+    GenericTransferInternal(direction, endpoint_number, buffer, length, timeout,
+                            callback, task_runner_);
+  } else {
     task_runner_->PostTask(
-        FROM_HERE, base::Bind(callback, USB_TRANSFER_DISCONNECT, nullptr, 0));
-    return;
-  }
-
-  uint8_t endpoint_address =
-      ConvertEndpointDirection(direction) | endpoint_number;
-  auto it = endpoints_.find(endpoint_address);
-  if (it == endpoints_.end()) {
-    USB_LOG(USER) << "Endpoint address " << static_cast<int>(endpoint_address)
-                  << " is not part of a claimed interface.";
-    task_runner_->PostTask(
-        FROM_HERE, base::Bind(callback, USB_TRANSFER_ERROR, nullptr, 0));
-    return;
-  }
-
-  std::unique_ptr<Transfer> transfer(new (0) Transfer(buffer, callback));
-  transfer->urb.endpoint = endpoint_address;
-  transfer->urb.buffer_length = length;
-  transfer->urb.type = ConvertTransferType(it->second.type);
-
-  // USBDEVFS_SUBMITURB appears to be non-blocking as completion is reported
-  // by USBDEVFS_REAPURBNDELAY. This code assumes a recent kernel that can
-  // accept arbitrarily large transfer requests, hopefully also using a scatter-
-  // gather list.
-  int rc = HANDLE_EINTR(ioctl(fd_.get(), USBDEVFS_SUBMITURB, &transfer->urb));
-  if (rc) {
-    rc = logging::GetLastSystemErrorCode();
-    USB_PLOG(DEBUG) << "Failed to submit transfer";
-    task_runner_->PostTask(
-        FROM_HERE, base::Bind(callback, ConvertTransferResult(rc), nullptr, 0));
-  } else {
-    SetUpTimeoutCallback(transfer.get(), timeout);
-    transfers_.push_back(std::move(transfer));
+        FROM_HERE,
+        base::Bind(&UsbDeviceHandleUsbfs::GenericTransferInternal, this,
+                   direction, endpoint_number, buffer, length, timeout,
+                   callback, base::ThreadTaskRunnerHandle::Get()));
   }
 }
 
 const UsbInterfaceDescriptor* UsbDeviceHandleUsbfs::FindInterfaceByEndpoint(
     uint8_t endpoint_address) {
   auto it = endpoints_.find(endpoint_address);
   if (it != endpoints_.end())
     return it->second.interface;
   return nullptr;
 }
@@ skipping 158 matching lines @@
   if (rc) {
     rc = logging::GetLastSystemErrorCode();
     USB_PLOG(DEBUG) << "Failed to submit transfer";
     ReportIsochronousError(packet_lengths, callback, ConvertTransferResult(rc));
   } else {
     SetUpTimeoutCallback(transfer.get(), timeout);
     transfers_.push_back(std::move(transfer));
   }
 }
 
+void UsbDeviceHandleUsbfs::GenericTransferInternal(
+    UsbEndpointDirection direction,
+    uint8_t endpoint_number,
+    scoped_refptr<net::IOBuffer> buffer,
+    size_t length,
+    unsigned int timeout,
+    const TransferCallback& callback,
+    scoped_refptr<base::SingleThreadTaskRunner> callback_runner) {
+  if (!device_) {
+    callback_runner->PostTask(
+        FROM_HERE, base::Bind(callback, USB_TRANSFER_DISCONNECT, nullptr, 0));
+    return;
+  }
+
+  uint8_t endpoint_address =
+      ConvertEndpointDirection(direction) | endpoint_number;
+  auto it = endpoints_.find(endpoint_address);
+  if (it == endpoints_.end()) {
+    USB_LOG(USER) << "Endpoint address " << static_cast<int>(endpoint_address)
+                  << " is not part of a claimed interface.";
+    callback_runner->PostTask(
+        FROM_HERE, base::Bind(callback, USB_TRANSFER_ERROR, nullptr, 0));
+    return;
+  }
+
+  std::unique_ptr<Transfer> transfer(
+      new (0) Transfer(buffer, callback, callback_runner));
+  transfer->urb.endpoint = endpoint_address;
+  transfer->urb.buffer_length = length;
+  transfer->urb.type = ConvertTransferType(it->second.type);
+
+  // USBDEVFS_SUBMITURB appears to be non-blocking as completion is reported
+  // by USBDEVFS_REAPURBNDELAY. This code assumes a recent kernel that can
+  // accept arbitrarily large transfer requests, hopefully also using a scatter-
+  // gather list.
+  int rc = HANDLE_EINTR(ioctl(fd_.get(), USBDEVFS_SUBMITURB, &transfer->urb));
+  if (rc) {
+    rc = logging::GetLastSystemErrorCode();
+    USB_PLOG(DEBUG) << "Failed to submit transfer";
+    callback_runner->PostTask(
+        FROM_HERE, base::Bind(callback, ConvertTransferResult(rc), nullptr, 0));
+  } else {
+    SetUpTimeoutCallback(transfer.get(), timeout);
+    transfers_.push_back(std::move(transfer));
+  }
+}
+
 void UsbDeviceHandleUsbfs::ReapedUrbs(const std::vector<usbdevfs_urb*>& urbs) {
   for (const auto& urb : urbs) {
     Transfer* this_transfer = static_cast<Transfer*>(urb->usercontext);
     DCHECK_EQ(urb, &this_transfer->urb);
     auto it = std::find_if(
         transfers_.begin(), transfers_.end(),
         [this_transfer](const std::unique_ptr<Transfer>& transfer) -> bool {
           return transfer.get() == this_transfer;
         });
     DCHECK(it != transfers_.end());
@@ skipping 16 matching lines @@
     std::vector<IsochronousPacket> packets(transfer->urb.number_of_packets);
     for (size_t i = 0; i < packets.size(); ++i) {
       packets[i].length = transfer->urb.iso_frame_desc[i].length;
       packets[i].transferred_length =
           transfer->urb.iso_frame_desc[i].actual_length;
       packets[i].status = ConvertTransferResult(
           transfer->urb.status == 0 ? transfer->urb.iso_frame_desc[i].status
                                     : transfer->urb.status);
     }
 
-    transfer->isoc_callback.Run(transfer->buffer, packets);
+    transfer->RunIsochronousCallback(packets);
   } else {
     if (transfer->urb.status == 0 &&
         transfer->urb.type == USBDEVFS_URB_TYPE_CONTROL) {
       // Copy the result of the control transfer back into the original buffer.
       memcpy(transfer->buffer->data(),
              transfer->control_transfer_buffer->data() + 8,
              transfer->urb.actual_length);
     }
 
-    transfer->callback.Run(ConvertTransferResult(-transfer->urb.status),
-                           transfer->buffer, transfer->urb.actual_length);
+    transfer->RunCallback(ConvertTransferResult(-transfer->urb.status),
+                          transfer->urb.actual_length);
   }
 }
 
 void UsbDeviceHandleUsbfs::RefreshEndpointInfo() {
   endpoints_.clear();
 
   const UsbConfigDescriptor* config = device_->GetActiveConfiguration();
   if (!config)
     return;
 
@@ skipping 48 matching lines @@
   // kernel and may be reaped later.
   transfer->cancelled = true;
   transfer->timeout_closure.Cancel();
   if (transfer->urb.type == USBDEVFS_URB_TYPE_ISO) {
     std::vector<IsochronousPacket> packets(transfer->urb.number_of_packets);
     for (size_t i = 0; i < packets.size(); ++i) {
       packets[i].length = transfer->urb.iso_frame_desc[i].length;
       packets[i].transferred_length = 0;
       packets[i].status = status;
     }
-    transfer->isoc_callback.Run(transfer->buffer, packets);
+    transfer->RunIsochronousCallback(packets);
   } else {
-    transfer->callback.Run(status, transfer->buffer, 0);
+    transfer->RunCallback(status, 0);
   }
 }
 
 }  // namespace device