usbdmx: Add support for having multiple inflight transfers

plugins/usbdmx/AVLdiyD512.cpp

@@ -124,7 +124,6 @@ class AVLdiyAsyncUsbSender : public AsyncUsbSender {
   }
 
   ~AVLdiyAsyncUsbSender() {
-    CancelTransfer();
     delete[] m_control_setup_buffer;
   }
 

plugins/usbdmx/AnymauDMX.cpp

@@ -124,7 +124,6 @@ class AnymaAsyncUsbSender : public AsyncUsbSender {
   }
 
   ~AnymaAsyncUsbSender() {
-    CancelTransfer();
     delete[] m_control_setup_buffer;
   }
 

plugins/usbdmx/AsyncUsbReceiver.cpp

@@ -28,13 +28,16 @@ namespace usbdmx {
 
 AsyncUsbReceiver::AsyncUsbReceiver(ola::usb::LibUsbAdaptor *adaptor,
                                    libusb_device *usb_device,
-                                   PluginAdaptor *plugin_adaptor)
-    : AsyncUsbTransceiverBase(adaptor, usb_device),
+                                   PluginAdaptor *plugin_adaptor,
+                                   unsigned int num_transfers)
+    : AsyncUsbTransceiverBase(adaptor, usb_device, num_transfers),
       m_plugin_adaptor(plugin_adaptor),
       m_inited_with_handle(false) {
 }
 
 AsyncUsbReceiver::~AsyncUsbReceiver() {
+  ola::thread::MutexLocker locker(&m_mutex);
+  CancelTransfer();
   if (!m_inited_with_handle) {
     m_adaptor->Close(m_usb_handle);
   }
@@ -58,25 +61,20 @@ bool AsyncUsbReceiver::Start() {
     return false;
   }
   ola::thread::MutexLocker locker(&m_mutex);
-  return PerformTransfer();
+  bool success = true;
+  while (!m_idle.empty() && 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;
   }

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 num_transfers maximum number of inflight transfers.
    */
   AsyncUsbReceiver(ola::usb::LibUsbAdaptor* const adaptor,
                    libusb_device *usb_device,
-                   PluginAdaptor *plugin_adaptor);
+                   PluginAdaptor *plugin_adaptor,
+                   unsigned int num_transfers = 1);
 
   /**
    * @brief Destructor

plugins/usbdmx/AsyncUsbSender.cpp

@@ -30,12 +30,15 @@ namespace usbdmx {
 using ola::usb::LibUsbAdaptor;
 
 AsyncUsbSender::AsyncUsbSender(LibUsbAdaptor *adaptor,
-                               libusb_device *usb_device)
-    : AsyncUsbTransceiverBase(adaptor, usb_device),
+                               libusb_device *usb_device,
+                               unsigned int num_transfers)
+    : AsyncUsbTransceiverBase(adaptor, usb_device, num_transfers),
       m_pending_tx(false) {
 }
 
 AsyncUsbSender::~AsyncUsbSender() {
+  ola::thread::MutexLocker locker(&m_mutex);
+  CancelTransfer();
   m_adaptor->Close(m_usb_handle);
 }
 
@@ -45,7 +48,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 +60,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);
   }

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 num_transfers maximum number of inflight transfers.
    */
   AsyncUsbSender(ola::usb::LibUsbAdaptor* const adaptor,
-                 libusb_device *usb_device);
+                 libusb_device *usb_device,
+                 unsigned int num_transfers = 1);
 
   /**
    * @brief Destructor

plugins/usbdmx/AsyncUsbTransceiverBase.cpp

@@ -18,6 +18,8 @@
  * Copyright (C) 2014 Simon Newton
  */
 
+#include <assert.h>
+
 #include "plugins/usbdmx/AsyncUsbTransceiverBase.h"
 
 #include "libs/usb/LibUsbAdaptor.h"
@@ -29,36 +31,70 @@ 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);
+
+  {
+    ola::thread::MutexLocker locker(&m_mutex);
+    if (m_inflight.empty()) {
+      m_cond.Signal();
+    }
+  }
 }
-}  // namespace
 
 AsyncUsbTransceiverBase::AsyncUsbTransceiverBase(LibUsbAdaptor *adaptor,
-                                                 libusb_device *usb_device)
+                                                 libusb_device *usb_device,
+                                                 unsigned int num_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 < num_transfers; i++) {
+    m_idle.push(m_adaptor->AllocTransfer(0));
+  }
   m_adaptor->RefDevice(usb_device);
 }
 
 AsyncUsbTransceiverBase::~AsyncUsbTransceiverBase() {
-  CancelTransfer();
+  ola::thread::MutexLocker locker(&m_mutex);
+
   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 +103,86 @@ bool AsyncUsbTransceiverBase::Init() {
 }
 
 void AsyncUsbTransceiverBase::CancelTransfer() {
-  if (!m_transfer) {
-    return;
+  m_suppress_continuation = true;
+
+  inflight_set::iterator i;
+  for (i = m_inflight.begin(); i != m_inflight.end(); ++i) {
+    int ret = m_adaptor->CancelTransfer(*i);
+    /* We demote NOT_FOUND errors to debug here because transfer
+     * cancellation naturally races with completion.
+     *
+     * This error simply means the transfer is not in the "in progress"
+     * state anymore.
+     *
+     * Even if we were to check for this before cancelling it could still
+     * happen that the transfer completes between the check and the
+     * cancellation taking effect.
+     */
+    if (ret != 0) {
+      log_level level = ret == LIBUSB_ERROR_NOT_FOUND
+          ? ola::OLA_LOG_DEBUG : ola::OLA_LOG_WARN;
+      OLA_LOG(level)
+          << "libusb_cancel_transfer returned "
+          << m_adaptor->ErrorCodeToString(ret);
+    }
   }
 
-  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;
-      }
-    }
+  while (!m_inflight.empty()) {
+    /* Wait for cancelled transfers to complete. Not waiting for the
+     * callbacks to happen is undefined behaviour according to libusb
+     * docs. */
+    m_cond.Wait(&m_mutex);
   }
+}
+
+struct libusb_transfer *AsyncUsbTransceiverBase::CurrentTransfer() {
+  /* This should never happen. In AsyncUsbSender::SendDMX we only ask the
+   * superclass to perform a transfer when m_idle is not empty and in
+   * AsyncUsbReceiver::Start we kick off the right number of transfers so
+   * at completion, when another one would be performed, m_idle will not be
+   * empty either. */
+  assert(!m_idle.empty());
 
-  m_suppress_continuation = false;
+  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