usb: removed handler thread, do work in get buffer calls

1 files changed, 75 insertions, 85 deletions

diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index 0fa856d34..4259c42ed 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -17,13 +17,9 @@
 
 #include "libusb1_base.hpp"
 #include <uhd/transport/usb_zero_copy.hpp>
-#include <uhd/transport/bounded_buffer.hpp>
 #include <uhd/transport/buffer_pool.hpp>
-#include <uhd/utils/thread_priority.hpp>
 #include <uhd/utils/msg.hpp>
-#include <uhd/utils/tasks.hpp>
 #include <uhd/exception.hpp>
-#include <boost/function.hpp>
 #include <boost/foreach.hpp>
 #include <boost/thread/thread.hpp>
 #include <list>
@@ -44,18 +40,9 @@ static const size_t DEFAULT_XFER_SIZE = 32*512; //bytes
  * to ensure that they are compiled with the same calling convention as libusb.
  */
 
-//! helper function: handles all rx async callbacks
-static void LIBUSB_CALL libusb_async_rx_cb(libusb_transfer *lut){
-    if(lut->actual_length == 0) {
-        UHD_ASSERT_THROW(libusb_submit_transfer(lut) == 0); //get out until you find some real data
-        return;
-    }
-    (*static_cast<boost::function<void()> *>(lut->user_data))();
-}
-
-//! helper function: handles all tx async callbacks
-static void LIBUSB_CALL libusb_async_tx_cb(libusb_transfer *lut) {
-    (*static_cast<boost::function<void()> *>(lut->user_data))();
+//! helper function: handles all async callbacks
+static void LIBUSB_CALL libusb_async_cb(libusb_transfer *lut){
+    *(static_cast<bool *>(lut->user_data)) = true;
 }
 
 //! callback to free transfer upon cancellation
@@ -64,6 +51,34 @@ static void LIBUSB_CALL cancel_transfer_cb(libusb_transfer *lut){
     else UHD_MSG(error) << "libusb cancel_transfer unexpected status " << lut->status << std::endl;
 }
 
+/*!
+ * Wait for a managed buffer to become complete.
+ *
+ * This routine processes async events until the transaction completes.
+ * We must call the libusb handle events in a loop because the handler
+ * may complete managed buffers other than the one we are waiting on.
+ *
+ * We cannot determine if handle events timed out or processed an event.
+ * Therefore, the timeout condition is handled by using boost system time.
+ *
+ * \param ctx the libusb context structure
+ * \param timeout the wait timeout in seconds
+ * \param completed a reference to the completed flag
+ * \return true for completion, false for timeout
+ */
+UHD_INLINE bool wait_for_completion(libusb_context *ctx, const double timeout, bool &completed){
+    const boost::system_time timeout_time = boost::get_system_time() + boost::posix_time::microseconds(long(timeout*1000000));
+
+    while (not completed and (boost::get_system_time() < timeout_time)){
+        timeval tv;
+        tv.tv_sec = 0;
+        tv.tv_usec = 10000; /*10ms*/
+        libusb_handle_events_timeout(ctx, &tv);
+    }
+
+    return completed;
+}
+
 /***********************************************************************
  * Reusable managed receiver buffer:
  *  - Associated with a particular libusb transfer struct.
@@ -72,23 +87,32 @@ static void LIBUSB_CALL cancel_transfer_cb(libusb_transfer *lut){
 class libusb_zero_copy_mrb : public managed_recv_buffer{
 public:
     libusb_zero_copy_mrb(libusb_transfer *lut):
-        _lut(lut), _expired(true) { /* NOP */ }
+        _ctx(libusb::session::get_global_session()->get_context()),
+        _lut(lut), _expired(false) { /* NOP */ }
 
     void release(void){
         if (_expired) return;
+        completed = false;
         UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
         _expired = true;
     }
 
-    sptr get_new(void){
-        _expired = false;
-        return make_managed_buffer(this);
+    sptr get_new(const double timeout, size_t &index){
+        if (wait_for_completion(_ctx, timeout, completed)){
+            index++;
+            _expired = false;
+            return make_managed_buffer(this);
+        }
+        return managed_recv_buffer::sptr();
     }
 
+    bool completed;
+
 private:
     const void *get_buff(void) const{return _lut->buffer;}
     size_t get_size(void) const{return _lut->actual_length;}
 
+    libusb_context *_ctx;
     libusb_transfer *_lut;
     bool _expired;
 };
@@ -101,25 +125,34 @@ private:
 class libusb_zero_copy_msb : public managed_send_buffer{
 public:
     libusb_zero_copy_msb(libusb_transfer *lut):
-        _lut(lut), _expired(true) { /* NOP */ }
+        _ctx(libusb::session::get_global_session()->get_context()),
+        _lut(lut), _expired(false) { /* NOP */ }
 
     void commit(size_t len){
         if (_expired) return;
+        completed = false;
         _lut->length = len;
-        if(len == 0) libusb_async_tx_cb(_lut);
+        if (len == 0) libusb_async_cb(_lut);
         else UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
         _expired = true;
     }
 
-    sptr get_new(void){
-        _expired = false;
-        return make_managed_buffer(this);
+    sptr get_new(const double timeout, size_t &index){
+        if (wait_for_completion(_ctx, timeout, completed)){
+            index++;
+            _expired = false;
+            return make_managed_buffer(this);
+        }
+        return managed_send_buffer::sptr();
     }
 
+    bool completed;
+
 private:
     void *get_buff(void) const{return _lut->buffer;}
     size_t get_size(void) const{return _lut->length;}
 
+    libusb_context *_ctx;
     libusb_transfer *_lut;
     bool _expired;
 };
@@ -143,8 +176,8 @@ public:
         _num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_XFERS))),
         _recv_buffer_pool(buffer_pool::make(_num_recv_frames, _recv_frame_size)),
         _send_buffer_pool(buffer_pool::make(_num_send_frames, _send_frame_size)),
-        _pending_recv_buffs(_num_recv_frames),
-        _pending_send_buffs(_num_send_frames)
+        _next_recv_buff_index(0),
+        _next_send_buff_index(0)
     {
         _handle->claim_interface(2 /*in interface*/);
         _handle->claim_interface(1 /*out interface*/);
@@ -155,10 +188,7 @@ public:
             libusb_transfer *lut = libusb_alloc_transfer(0);
             UHD_ASSERT_THROW(lut != NULL);
 
-            _mrb_pool.push_back(libusb_zero_copy_mrb(lut));
-            _callbacks.push_back(boost::bind(
-                &libusb_zero_copy_impl::handle_recv, this, &_mrb_pool.back()
-            ));
+            _mrb_pool.push_back(boost::shared_ptr<libusb_zero_copy_mrb>(new libusb_zero_copy_mrb(lut)));
 
             libusb_fill_bulk_transfer(
                 lut,                                                    // transfer
@@ -166,13 +196,13 @@ public:
                 (recv_endpoint & 0x7f) | 0x80,                          // endpoint
                 static_cast<unsigned char *>(_recv_buffer_pool->at(i)), // buffer
                 this->get_recv_frame_size(),                            // length
-                libusb_transfer_cb_fn(&libusb_async_rx_cb),             // callback
-                static_cast<void *>(&_callbacks.back()),                // user_data
+                libusb_transfer_cb_fn(&libusb_async_cb),                // callback
+                static_cast<void *>(&_mrb_pool.back()->completed),      // user_data
                 0                                                       // timeout (ms)
             );
 
             _all_luts.push_back(lut);
-            _mrb_pool.back().get_new();
+            _mrb_pool.back()->release();
         }
 
         //allocate libusb transfer structs and managed send buffers
@@ -181,10 +211,7 @@ public:
             libusb_transfer *lut = libusb_alloc_transfer(0);
             UHD_ASSERT_THROW(lut != NULL);
 
-            _msb_pool.push_back(libusb_zero_copy_msb(lut));
-            _callbacks.push_back(boost::bind(
-                &libusb_zero_copy_impl::handle_send, this, &_msb_pool.back()
-            ));
+            _msb_pool.push_back(boost::shared_ptr<libusb_zero_copy_msb>(new libusb_zero_copy_msb(lut)));
 
             libusb_fill_bulk_transfer(
                 lut,                                                    // transfer
@@ -192,20 +219,14 @@ public:
                 (send_endpoint & 0x7f) | 0x00,                          // endpoint
                 static_cast<unsigned char *>(_send_buffer_pool->at(i)), // buffer
                 this->get_send_frame_size(),                            // length
-                libusb_transfer_cb_fn(&libusb_async_tx_cb),             // callback
-                static_cast<void *>(&_callbacks.back()),                // user_data
+                libusb_transfer_cb_fn(&libusb_async_cb),                // callback
+                static_cast<void *>(&_msb_pool.back()->completed),      // user_data
                 0                                                       // timeout
             );
 
             _all_luts.push_back(lut);
-            libusb_async_tx_cb(lut);
+            _msb_pool.back()->commit(0);
         }
-
-        //spawn the event handler threads
-        const size_t concurrency = hints.cast<size_t>("concurrency_hint", 1);
-        for (size_t i = 0; i < concurrency; i++) _event_loop_tasks.push_back(task::make(
-            boost::bind(&libusb_zero_copy_impl::run_event_loop, this)
-        ));
     }
 
     ~libusb_zero_copy_impl(void){
@@ -222,19 +243,13 @@ public:
     }
 
     managed_recv_buffer::sptr get_recv_buff(double timeout){
-        libusb_zero_copy_mrb *mrb = NULL;
-        if (_pending_recv_buffs.pop_with_timed_wait(mrb, timeout)){
-            return mrb->get_new();
-        }
-        return managed_recv_buffer::sptr();
+        if (_next_recv_buff_index == _num_recv_frames) _next_recv_buff_index = 0;
+        return _mrb_pool[_next_recv_buff_index]->get_new(timeout, _next_recv_buff_index);
     }
 
     managed_send_buffer::sptr get_send_buff(double timeout){
-        libusb_zero_copy_msb *msb = NULL;
-        if (_pending_send_buffs.pop_with_timed_wait(msb, timeout)){
-            return msb->get_new();
-        }
-        return managed_send_buffer::sptr();
+        if (_next_send_buff_index == _num_send_frames) _next_send_buff_index = 0;
+        return _msb_pool[_next_send_buff_index]->get_new(timeout, _next_send_buff_index);
     }
 
     size_t get_num_recv_frames(void) const { return _num_recv_frames; }
@@ -244,44 +259,19 @@ public:
     size_t get_send_frame_size(void) const { return _send_frame_size; }
 
 private:
-    //! Handle a bound async callback for recv
-    void handle_recv(libusb_zero_copy_mrb *mrb){
-        _pending_recv_buffs.push_with_haste(mrb);
-    }
-
-    //! Handle a bound async callback for send
-    void handle_send(libusb_zero_copy_msb *msb){
-        _pending_send_buffs.push_with_haste(msb);
-    }
-
     libusb::device_handle::sptr _handle;
     const size_t _recv_frame_size, _num_recv_frames;
     const size_t _send_frame_size, _num_send_frames;
 
     //! Storage for transfer related objects
     buffer_pool::sptr _recv_buffer_pool, _send_buffer_pool;
-    bounded_buffer<libusb_zero_copy_mrb *> _pending_recv_buffs;
-    bounded_buffer<libusb_zero_copy_msb *> _pending_send_buffs;
-    std::list<libusb_zero_copy_mrb> _mrb_pool;
-    std::list<libusb_zero_copy_msb> _msb_pool;
-    std::list<boost::function<void()> > _callbacks;
+    std::vector<boost::shared_ptr<libusb_zero_copy_mrb> > _mrb_pool;
+    std::vector<boost::shared_ptr<libusb_zero_copy_msb> > _msb_pool;
+    size_t _next_recv_buff_index, _next_send_buff_index;
 
     //! a list of all transfer structs we allocated
     std::list<libusb_transfer *> _all_luts;
 
-    //! event handler threads
-    std::list<task::sptr> _event_loop_tasks;
-
-    void run_event_loop(void){
-        set_thread_priority_safe();
-        libusb_context *context = libusb::session::get_global_session()->get_context();
-        while (not boost::this_thread::interruption_requested()){
-            timeval tv;
-            tv.tv_sec = 0;
-            tv.tv_usec = 100000; //100ms
-            libusb_handle_events_timeout(context, &tv);
-        }
-    }
 
 };