usb: moved event handler thread into the zero copy interface

3 files changed, 29 insertions, 18 deletions

diff --git a/host/docs/transport.rst b/host/docs/transport.rst
index d6a146c67..30fc1d78f 100644
--- a/host/docs/transport.rst
+++ b/host/docs/transport.rst
@@ -79,3 +79,4 @@ The following parameters can be used to alter the transport's default behavior:
 * **num_recv_frames:** The number of simultaneous receive transfers
 * **send_frame_size:** The size of a single send transfers in bytes
 * **num_send_frames:** The number of simultaneous send transfers
+* **concurrency_hint:** The number of threads to run the event handler
diff --git a/host/lib/transport/libusb1_base.cpp b/host/lib/transport/libusb1_base.cpp
index 910b04fc8..cfa77d9ca 100644
--- a/host/lib/transport/libusb1_base.cpp
+++ b/host/lib/transport/libusb1_base.cpp
@@ -16,12 +16,10 @@
 //
 
 #include "libusb1_base.hpp"
-#include <uhd/utils/thread_priority.hpp>
 #include <uhd/utils/assert.hpp>
 #include <uhd/types/dict.hpp>
 #include <boost/weak_ptr.hpp>
 #include <boost/foreach.hpp>
-#include <boost/thread.hpp>
 #include <iostream>
 
 using namespace uhd;
@@ -35,12 +33,9 @@ public:
     libusb_session_impl(void){
         UHD_ASSERT_THROW(libusb_init(&_context) == 0);
         libusb_set_debug(_context, debug_level);
-        _thread_group.create_thread(boost::bind(&libusb_session_impl::run_event_loop, this));
     }
 
     ~libusb_session_impl(void){
-        _running = false;
-        _thread_group.join_all();
         libusb_exit(_context);
     }
 
@@ -50,19 +45,6 @@ public:
 
 private:
     libusb_context *_context;
-    boost::thread_group _thread_group;
-    bool _running;
-
-    void run_event_loop(void){
-        set_thread_priority_safe();
-        _running = true;
-        timeval tv;
-        while(_running){
-            tv.tv_sec = 0;
-            tv.tv_usec = 100000; //100ms
-            libusb_handle_events_timeout(this->get_context(), &tv);
-        }
-    }
 };
 
 libusb::session::sptr libusb::session::get_global_session(void){
diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index df6db1eb9..f589d7c77 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -18,6 +18,7 @@
 #include "libusb1_base.hpp"
 #include <uhd/transport/usb_zero_copy.hpp>
 #include <uhd/transport/bounded_buffer.hpp>
+#include <uhd/utils/thread_priority.hpp>
 #include <uhd/utils/assert.hpp>
 #include <boost/shared_array.hpp>
 #include <boost/foreach.hpp>
@@ -290,6 +291,11 @@ public:
         const device_addr_t &hints
     );
 
+    ~libusb_zero_copy_impl(void){
+        _threads_running = false;
+        _thread_group.join_all();
+    }
+
     managed_recv_buffer::sptr get_recv_buff(double);
     managed_send_buffer::sptr get_send_buff(double);
 
@@ -318,6 +324,22 @@ private:
     const size_t _recv_frame_size, _num_recv_frames;
     const size_t _send_frame_size, _num_send_frames;
     usb_endpoint::sptr _recv_ep, _send_ep;
+
+    //event handler threads
+    boost::thread_group _thread_group;
+    bool _threads_running;
+
+    void run_event_loop(void){
+        set_thread_priority_safe();
+        libusb::session::sptr session = libusb::session::get_global_session();
+        _threads_running = true;
+        while(_threads_running){
+            timeval tv;
+            tv.tv_sec = 0;
+            tv.tv_usec = 100000; //100ms
+            libusb_handle_events_timeout(session->get_context(), &tv);
+        }
+    }
 };
 
 /*
@@ -355,6 +377,12 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(
                               this->get_send_frame_size(),  // buffer size per transfer
                               this->get_num_send_frames()   // number of libusb transfers
     ));
+
+    //spawn the event handler threads
+    size_t concurrency = hints.cast<size_t>("concurrency_hint", 1);
+    for (size_t i = 0; i < concurrency; i++) _thread_group.create_thread(
+        boost::bind(&libusb_zero_copy_impl::run_event_loop, this)
+    );
 }
 
 /*