usbdmx: Add support for having multiple inflight transfers

DanielG · DanielG · commit 51918fcf4e31 · 2020-04-06T13:41:07.000+02:00
Currently we only use a single active transfer for both sending and
reveiving usb packets.

This is suboptimal in terms of latency, USB actually guarantees certain
turnaround times for interrupt transfers for example but that only really
works if there is always an IRP ready to go in the USB host controller.

According to my measurments interrupt transfers with bInterval=1 can be
delayed by some tens of milliseconds instead of happening evey millisecond
on the dot if there are periods with no transfers ready to go.

In order to support multiple transfers being inflight I replace the
transfer state machinery in AsyncUsbTransceiverBase with a set of inflight
transfers and a queue of idle transfers.

The bookkeeping happens in a new TransferCompleteInternal method which
wraps the TransferComplete() method provided by subclasses. This split also
allowed moving some duplicated logic for LIBUSB_TRANSFER_NO_DEVICE to the
superclass.

The number of transfers to allow is passed to AsyncUsb{Sender,Receiver} as
a constructor argument since supporting more than one needs some
cooperation from the derived classes. Currently everything just defaults to
one transfer which shouldn't change the behaviour one bit.
diff --git a/plugins/usbdmx/AsyncUsbReceiver.cpp b/plugins/usbdmx/AsyncUsbReceiver.cpp
@@ -28,8 +28,9 @@ namespace usbdmx {
 
 AsyncUsbReceiver::AsyncUsbReceiver(ola::usb::LibUsbAdaptor *adaptor,
                                    libusb_device *usb_device,
-                                   PluginAdaptor *plugin_adaptor)
-    : AsyncUsbTransceiverBase(adaptor, usb_device),
+                                   PluginAdaptor *plugin_adaptor,
+                                   unsigned int n_transfers)
+    : AsyncUsbTransceiverBase(adaptor, usb_device, n_transfers),
       m_plugin_adaptor(plugin_adaptor),
       m_inited_with_handle(false) {
 }
@@ -58,25 +59,20 @@ bool AsyncUsbReceiver::Start() {
     return false;
   }
   ola::thread::MutexLocker locker(&m_mutex);
-  return PerformTransfer();
+  bool success = true;
+  while(!m_idle.empty() && success) {
+    success = success && PerformTransfer();
+  }
+  return success;
 }
 
 void AsyncUsbReceiver::TransferComplete(struct libusb_transfer *transfer) {
-  if (transfer != m_transfer) {
-    OLA_WARN << "Mismatched libusb transfer: " << transfer << " != "
-             << m_transfer;
-    return;
-  }
-
   if (transfer->status != LIBUSB_TRANSFER_COMPLETED &&
       transfer->status != LIBUSB_TRANSFER_TIMED_OUT ) {
     OLA_WARN << "Transfer returned " << transfer->status;
   }
 
   ola::thread::MutexLocker locker(&m_mutex);
-  m_transfer_state = (transfer->status == LIBUSB_TRANSFER_NO_DEVICE ?
-      DISCONNECTED : IDLE);
-
   if (m_suppress_continuation) {
     return;
   }
diff --git a/plugins/usbdmx/AsyncUsbReceiver.h b/plugins/usbdmx/AsyncUsbReceiver.h
@@ -50,10 +50,12 @@ class AsyncUsbReceiver: public AsyncUsbTransceiverBase {
    * @param adaptor the LibUsbAdaptor to use.
    * @param usb_device the libusb_device to use for the widget.
    * @param plugin_adaptor the PluginAdaptor to use for the widget.
+   * @param n_transfers maximum number of inflight transfers.
    */
   AsyncUsbReceiver(ola::usb::LibUsbAdaptor* const adaptor,
                    libusb_device *usb_device,
-                   PluginAdaptor *plugin_adaptor);
+                   PluginAdaptor *plugin_adaptor,
+                   unsigned int n_transfers = 1);
 
   /**
    * @brief Destructor
diff --git a/plugins/usbdmx/AsyncUsbSender.cpp b/plugins/usbdmx/AsyncUsbSender.cpp
@@ -30,8 +30,9 @@ namespace usbdmx {
 using ola::usb::LibUsbAdaptor;
 
 AsyncUsbSender::AsyncUsbSender(LibUsbAdaptor *adaptor,
-                               libusb_device *usb_device)
-    : AsyncUsbTransceiverBase(adaptor, usb_device),
+                               libusb_device *usb_device,
+                               unsigned int n_transfers)
+    : AsyncUsbTransceiverBase(adaptor, usb_device, n_transfers),
       m_pending_tx(false) {
 }
 
@@ -45,7 +46,7 @@ bool AsyncUsbSender::SendDMX(const DmxBuffer &buffer) {
     return false;
   }
   ola::thread::MutexLocker locker(&m_mutex);
-  if (m_transfer_state == IDLE) {
+  if (!m_idle.empty()) {
     PerformTransfer(buffer);
   } else {
     // Buffer incoming data so we can send it when the outstanding transfers
@@ -57,28 +58,19 @@ bool AsyncUsbSender::SendDMX(const DmxBuffer &buffer) {
 }
 
 void AsyncUsbSender::TransferComplete(struct libusb_transfer *transfer) {
-  if (transfer != m_transfer) {
-    OLA_WARN << "Mismatched libusb transfer: " << transfer << " != "
-             << m_transfer;
-    return;
-  }
-
   if (transfer->status != LIBUSB_TRANSFER_COMPLETED) {
     OLA_WARN << "Transfer returned "
              << m_adaptor->ErrorCodeToString(transfer->status);
   }
 
   ola::thread::MutexLocker locker(&m_mutex);
-  m_transfer_state = (transfer->status == LIBUSB_TRANSFER_NO_DEVICE ?
-      DISCONNECTED : IDLE);
-
   if (m_suppress_continuation) {
     return;
   }
 
   PostTransferHook();
 
-  if ((m_transfer_state == IDLE) && m_pending_tx) {
+  if (!m_device_disconnected && m_pending_tx) {
     m_pending_tx = false;
     PerformTransfer(m_tx_buffer);
   }
diff --git a/plugins/usbdmx/AsyncUsbSender.h b/plugins/usbdmx/AsyncUsbSender.h
@@ -45,9 +45,11 @@ class AsyncUsbSender: public AsyncUsbTransceiverBase {
    * @brief Create a new AsyncUsbSender.
    * @param adaptor the LibUsbAdaptor to use.
    * @param usb_device the libusb_device to use for the widget.
+   * @param n_transfers maximum number of inflight transfers.
    */
   AsyncUsbSender(ola::usb::LibUsbAdaptor* const adaptor,
-                 libusb_device *usb_device);
+                 libusb_device *usb_device,
+                 unsigned int n_transfers = 1);
 
   /**
    * @brief Destructor
diff --git a/plugins/usbdmx/AsyncUsbTransceiverBase.cpp b/plugins/usbdmx/AsyncUsbTransceiverBase.cpp
@@ -18,6 +18,8 @@
  * Copyright (C) 2014 Simon Newton
  */
 
+#include <stdexcept>
+
 #include "plugins/usbdmx/AsyncUsbTransceiverBase.h"
 
 #include "libs/usb/LibUsbAdaptor.h"
@@ -29,36 +31,64 @@ namespace usbdmx {
 
 using ola::usb::LibUsbAdaptor;
 
-namespace {
-
 /*
  * Called by libusb when the transfer completes.
  */
 #ifdef _WIN32
 __attribute__((__stdcall__))
 #endif  // _WIN32
-void AsyncCallback(struct libusb_transfer *transfer) {
+void AsyncTransferCallback(struct libusb_transfer *transfer) {
   AsyncUsbTransceiverBase *widget = reinterpret_cast<AsyncUsbTransceiverBase*>(
     transfer->user_data);
-  widget->TransferComplete(transfer);
+  widget->TransferCompleteInternal(transfer);
+}
+
+void AsyncUsbTransceiverBase::TransferCompleteInternal(
+  struct libusb_transfer *transfer) {
+  {
+    ola::thread::MutexLocker locker(&m_mutex);
+    inflight_set::iterator i = m_inflight.find(transfer);
+    if (i == m_inflight.end()) { // not found
+      OLA_WARN << "Mismatched libusb transfer: "
+               << transfer << " not in inflight set";
+      return;
+    }
+
+    m_inflight.erase(i);
+    m_idle.push(transfer);
+  }
+
+  if (transfer->status == LIBUSB_TRANSFER_NO_DEVICE)
+    m_device_disconnected = true;
+
+  TransferComplete(transfer);
 }
-}  // namespace
 
 AsyncUsbTransceiverBase::AsyncUsbTransceiverBase(LibUsbAdaptor *adaptor,
-                                                 libusb_device *usb_device)
+                                                 libusb_device *usb_device,
+                                                 unsigned int n_transfers)
     : m_adaptor(adaptor),
       m_usb_device(usb_device),
       m_usb_handle(NULL),
       m_suppress_continuation(false),
-      m_transfer_state(IDLE) {
-  m_transfer = m_adaptor->AllocTransfer(0);
+      m_device_disconnected(false) {
+  for (unsigned int i=0; i < n_transfers; i++) {
+    m_idle.push(m_adaptor->AllocTransfer(0));
+  }
   m_adaptor->RefDevice(usb_device);
 }
 
 AsyncUsbTransceiverBase::~AsyncUsbTransceiverBase() {
+  ola::thread::MutexLocker locker(&m_mutex);
+
   CancelTransfer();
+
   m_adaptor->UnrefDevice(m_usb_device);
-  m_adaptor->FreeTransfer(m_transfer);
+
+  while (!m_idle.empty()) {
+    m_adaptor->FreeTransfer(m_idle.front());
+    m_idle.pop();
+  }
 }
 
 bool AsyncUsbTransceiverBase::Init() {
@@ -67,62 +97,63 @@ bool AsyncUsbTransceiverBase::Init() {
 }
 
 void AsyncUsbTransceiverBase::CancelTransfer() {
-  if (!m_transfer) {
-    return;
-  }
-
-  bool canceled = false;
-  while (1) {
-    ola::thread::MutexLocker locker(&m_mutex);
-    if (m_transfer_state == IDLE || m_transfer_state == DISCONNECTED) {
-      break;
-    }
-    if (!canceled) {
-      m_suppress_continuation = true;
-      if (m_adaptor->CancelTransfer(m_transfer) == 0) {
-        canceled = true;
-      } else {
-        break;
-      }
+  m_suppress_continuation = true;
+
+  inflight_set::iterator i;
+  for (i = m_inflight.begin(); i != m_inflight.end(); ++i)
+  {
+    int ret = m_adaptor->CancelTransfer(*i);
+    if (ret != 0 && ret != LIBUSB_ERROR_NOT_FOUND) {
+      OLA_WARN << "libusb_cancel_transfer returned "
+               << m_adaptor->ErrorCodeToString(ret);
     }
   }
+}
 
-  m_suppress_continuation = false;
+struct libusb_transfer *AsyncUsbTransceiverBase::CurrentTransfer() {
+  if (m_idle.empty())
+    throw std::out_of_range("AsyncUsbTransceiverBase::m_idle is empty");
+
+  return m_idle.front();
 }
 
+
 void AsyncUsbTransceiverBase::FillControlTransfer(unsigned char *buffer,
                                                   unsigned int timeout) {
-  m_adaptor->FillControlTransfer(m_transfer, m_usb_handle, buffer,
-                                 &AsyncCallback, this, timeout);
+  m_adaptor->FillControlTransfer(CurrentTransfer(), m_usb_handle, buffer,
+                                 &AsyncTransferCallback, this, timeout);
 }
 
 void AsyncUsbTransceiverBase::FillBulkTransfer(unsigned char endpoint,
                                                unsigned char *buffer,
                                                int length,
                                                unsigned int timeout) {
-  m_adaptor->FillBulkTransfer(m_transfer, m_usb_handle, endpoint, buffer,
-                              length, &AsyncCallback, this, timeout);
+  m_adaptor->FillBulkTransfer(CurrentTransfer(), m_usb_handle, endpoint, buffer,
+                              length, &AsyncTransferCallback, this, timeout);
 }
 
 void AsyncUsbTransceiverBase::FillInterruptTransfer(unsigned char endpoint,
                                                     unsigned char *buffer,
                                                     int length,
                                                     unsigned int timeout) {
-  m_adaptor->FillInterruptTransfer(m_transfer, m_usb_handle, endpoint, buffer,
-                                   length, &AsyncCallback, this, timeout);
+  m_adaptor->FillInterruptTransfer(CurrentTransfer(), m_usb_handle, endpoint,
+                                   buffer, length, &AsyncTransferCallback,
+                                   this, timeout);
 }
 
 int AsyncUsbTransceiverBase::SubmitTransfer() {
-  int ret = m_adaptor->SubmitTransfer(m_transfer);
+  struct libusb_transfer *transfer = CurrentTransfer();
+  int ret = m_adaptor->SubmitTransfer(transfer);
   if (ret) {
     OLA_WARN << "libusb_submit_transfer returned "
              << m_adaptor->ErrorCodeToString(ret);
     if (ret == LIBUSB_ERROR_NO_DEVICE) {
-      m_transfer_state = DISCONNECTED;
+      m_device_disconnected = true;
     }
     return false;
   }
-  m_transfer_state = IN_PROGRESS;
+  m_inflight.insert(transfer);
+  m_idle.pop();
   return ret;
 }
 }  // namespace usbdmx
diff --git a/plugins/usbdmx/AsyncUsbTransceiverBase.h b/plugins/usbdmx/AsyncUsbTransceiverBase.h
@@ -22,6 +22,8 @@
 #define PLUGINS_USBDMX_ASYNCUSBTRANSCEIVERBASE_H_
 
 #include <libusb.h>
+#include <queue>
+#include <set>
 
 #include "libs/usb/LibUsbAdaptor.h"
 #include "ola/DmxBuffer.h"
@@ -42,9 +44,11 @@ class AsyncUsbTransceiverBase {
    * @brief Create a new AsyncUsbTransceiverBase.
    * @param adaptor the LibUsbAdaptor to use.
    * @param usb_device the libusb_device to use for the widget.
+   * @param n_transfers maximum number of inflight transfers.
    */
   AsyncUsbTransceiverBase(ola::usb::LibUsbAdaptor* const adaptor,
-                          libusb_device *usb_device);
+                          libusb_device *usb_device,
+                          unsigned int n_transfers);
 
   /**
    * @brief Destructor
@@ -126,20 +130,27 @@ class AsyncUsbTransceiverBase {
    */
   int SubmitTransfer();
 
-  enum TransferState {
-    IDLE,
-    IN_PROGRESS,
-    DISCONNECTED,
-  };
-
   libusb_device_handle *m_usb_handle;
   bool m_suppress_continuation;
-  struct libusb_transfer *m_transfer;
+  bool m_device_disconnected;
+
+  typedef std::set<struct libusb_transfer*> inflight_set;
+  inflight_set m_inflight;  // GUARDED_BY(m_mutex);
+  std::queue<struct libusb_transfer*> m_idle;  // GUARDED_BY(m_mutex);
 
-  TransferState m_transfer_state;  // GUARDED_BY(m_mutex);
   ola::thread::Mutex m_mutex;
 
  private:
+  /**
+   * @brief Called from the libusb callback when the asynchronous transfer
+   *   completes.
+   * @param transfer the completed transfer.
+   */
+  void TransferCompleteInternal(struct libusb_transfer *transfer);
+  friend void AsyncTransferCallback(struct libusb_transfer *transfer);
+
+  struct libusb_transfer *CurrentTransfer();
+
   DISALLOW_COPY_AND_ASSIGN(AsyncUsbTransceiverBase);
 };
 }  // namespace usbdmx
