Merge branch 'timeout' into usrp_e_mmap

1 files changed, 63 insertions, 149 deletions

diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index f9beb0b4c..ab48e4fc4 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -24,12 +24,10 @@
 #include <boost/thread.hpp>
 #include <vector>
 #include <iostream>
-#include <iomanip>
 
 using namespace uhd::transport;
 
-const int libusb_timeout = 0;
-
+static const double CLEANUP_TIMEOUT   = 0.2;    //seconds
 static const size_t DEFAULT_NUM_XFERS = 16;     //num xfers
 static const size_t DEFAULT_XFER_SIZE = 32*512; //bytes
 
@@ -84,10 +82,10 @@ public:
      * Get an available transfer:
      * For inputs, this is a just filled transfer.
      * For outputs, this is a just emptied transfer.
-     * \param timeout_ms the timeout to wait for a lut
+     * \param timeout the timeout to wait for a lut
      * \return the transfer pointer or NULL if timeout
      */
-    libusb_transfer *get_lut_with_wait(size_t timeout_ms = 100);
+    libusb_transfer *get_lut_with_wait(double timeout);
 
     //Callback use only
     void callback_handle_transfer(libusb_transfer *lut);
@@ -97,21 +95,18 @@ private:
     int  _endpoint;
     bool _input;
 
-    size_t _transfer_size;
-    size_t _num_transfers;
-
     //! hold a bounded buffer of completed transfers
     typedef bounded_buffer<libusb_transfer *> lut_buff_type;
     lut_buff_type::sptr _completed_list;
 
     //! a list of all transfer structs we allocated
-    std::vector<libusb_transfer *>  _all_luts;
+    std::vector<libusb_transfer *> _all_luts;
 
-    //! a list of shared arrays for the transfer buffers
-    std::vector<boost::shared_array<boost::uint8_t> > _buffers;
+    //! a block of memory for the transfer buffers
+    boost::shared_array<char> _buffer;
 
     // Calls for processing asynchronous I/O
-    libusb_transfer *allocate_transfer(int buff_len);
+    libusb_transfer *allocate_transfer(void *mem, size_t len);
     void print_transfer_status(libusb_transfer *lut);
 };
 
@@ -156,14 +151,12 @@ usb_endpoint::usb_endpoint(
 ):
     _handle(handle),
     _endpoint(endpoint),
-    _input(input),
-    _transfer_size(transfer_size),
-    _num_transfers(num_transfers)
+    _input(input)
 {
     _completed_list = lut_buff_type::make(num_transfers);
-
-    for (size_t i = 0; i < _num_transfers; i++){
-        _all_luts.push_back(allocate_transfer(_transfer_size));
+    _buffer = boost::shared_array<char>(new char[num_transfers*transfer_size]);
+    for (size_t i = 0; i < num_transfers; i++){
+        _all_luts.push_back(allocate_transfer(_buffer.get() + i*transfer_size, transfer_size));
 
         //input luts are immediately submitted to be filled
         //output luts go into the completed list as free buffers
@@ -187,7 +180,7 @@ usb_endpoint::~usb_endpoint(void){
     }
 
     //collect canceled transfers (drain the queue)
-    while (this->get_lut_with_wait() != NULL){};
+    while (this->get_lut_with_wait(CLEANUP_TIMEOUT) != NULL){};
 
     //free all transfers
     BOOST_FOREACH(libusb_transfer *lut, _all_luts){
@@ -200,23 +193,23 @@ usb_endpoint::~usb_endpoint(void){
  * Allocate a libusb transfer
  * The allocated transfer - and buffer it contains - is repeatedly
  * submitted, reaped, and reused and should not be freed until shutdown.
- * \param buff_len size of the individual buffer held by each transfer
+ * \param mem a pointer to the buffer memory
+ * \param len size of the individual buffer
  * \return pointer to an allocated libusb_transfer
  */
-libusb_transfer *usb_endpoint::allocate_transfer(int buff_len){
+libusb_transfer *usb_endpoint::allocate_transfer(void *mem, size_t len){
     libusb_transfer *lut = libusb_alloc_transfer(0);
-
-    boost::shared_array<boost::uint8_t> buff(new boost::uint8_t[buff_len]);
-    _buffers.push_back(buff); //store a reference to this shared array
+    UHD_ASSERT_THROW(lut != NULL);
 
     unsigned int endpoint = ((_endpoint & 0x7f) | (_input ? 0x80 : 0));
+    unsigned char *buff = reinterpret_cast<unsigned char *>(mem);
     libusb_transfer_cb_fn lut_callback = libusb_transfer_cb_fn(&callback);
 
     libusb_fill_bulk_transfer(lut,                // transfer
                               _handle->get(),     // dev_handle
                               endpoint,           // endpoint
-                              buff.get(),         // buffer
-                              buff_len,           // length
+                              buff,               // buffer
+                              len,                // length
                               lut_callback,       // callback
                               this,               // user_data
                               0);                 // timeout
@@ -239,6 +232,7 @@ void usb_endpoint::submit(libusb_transfer *lut){
  * \param lut pointer to an libusb_transfer
  */
 void usb_endpoint::print_transfer_status(libusb_transfer *lut){
+    std::cout << "here " << lut->status << std::endl;
     switch (lut->status) {
     case LIBUSB_TRANSFER_COMPLETED:
         if (lut->actual_length < lut->length) {
@@ -274,136 +268,52 @@ void usb_endpoint::print_transfer_status(libusb_transfer *lut){
     }
 }
 
-libusb_transfer *usb_endpoint::get_lut_with_wait(size_t timeout_ms){
+libusb_transfer *usb_endpoint::get_lut_with_wait(double timeout){
     boost::this_thread::disable_interruption di; //disable because the wait can throw
     libusb_transfer *lut;
-    if (_completed_list->pop_with_timed_wait(
-        lut, boost::posix_time::milliseconds(timeout_ms)
-    )) return lut;
+    if (_completed_list->pop_with_timed_wait(lut, timeout)) return lut;
     return NULL;
 }
 
 /***********************************************************************
- * Managed buffers
+ * USB zero_copy device class
  **********************************************************************/
-/*
- * Libusb managed receive buffer
- * Construct a recv buffer from a libusb transfer. The memory held by
- * the libusb transfer is exposed through the managed buffer interface.
- * Upon destruction, the transfer and buffer are resubmitted to the
- * endpoint for further use.
- */
-class libusb_managed_recv_buffer_impl : public managed_recv_buffer {
+class libusb_zero_copy_impl : public usb_zero_copy, public boost::enable_shared_from_this<libusb_zero_copy_impl> {
 public:
-    libusb_managed_recv_buffer_impl(libusb_transfer *lut,
-                                    usb_endpoint::sptr endpoint)
-        : _buff(lut->buffer, lut->length)
-    {
-        _lut = lut;
-        _endpoint = endpoint;
-    }
-
-    ~libusb_managed_recv_buffer_impl(void){
-        _endpoint->submit(_lut);
-    }
+    typedef boost::shared_ptr<libusb_zero_copy_impl> sptr;
 
-private:
-    const boost::asio::const_buffer &get(void) const{
-        return _buff;
-    }
+    libusb_zero_copy_impl(
+        libusb::device_handle::sptr handle,
+        unsigned int recv_endpoint, unsigned int send_endpoint,
+        size_t recv_xfer_size, size_t recv_num_xfers,
+        size_t send_xfer_size, size_t send_num_xfers
+    );
 
-    libusb_transfer *_lut;
-    usb_endpoint::sptr _endpoint;
-    const boost::asio::const_buffer _buff;
-};
+    managed_recv_buffer::sptr get_recv_buff(double);
+    managed_send_buffer::sptr get_send_buff(double);
 
-/*
- * Libusb managed send buffer
- * Construct a send buffer from a libusb transfer. The memory held by
- * the libusb transfer is exposed through the managed buffer interface.
- * Committing the buffer will set the data length and submit the buffer
- * to the endpoint. Submitting a buffer multiple times or destroying
- * the buffer before committing is an error. For the latter, the transfer
- * is returned to the endpoint with no data for reuse.
- */
-class libusb_managed_send_buffer_impl : public managed_send_buffer {
-public:
-    libusb_managed_send_buffer_impl(libusb_transfer *lut,
-                                    usb_endpoint::sptr endpoint)
-        : _buff(lut->buffer, lut->length), _committed(false)
-    {
-        _lut = lut;
-        _endpoint = endpoint;
-    }
+    size_t get_num_recv_frames(void) const { return _recv_num_frames; }
+    size_t get_num_send_frames(void) const { return _send_num_frames; }
 
-    ~libusb_managed_send_buffer_impl(void){
-        if (!_committed) {
-            _lut->length = 0;
-            _lut->actual_length = 0;
-            _endpoint->submit(_lut);
-        }
+private:
+    void release(libusb_transfer *lut){
+        _recv_ep->submit(lut);
     }
 
-    ssize_t commit(size_t num_bytes)
-    {
-        if (_committed) {
-            std::cerr << "UHD: send buffer already committed" << std::endl;
-            return 0;
-        }
-
-        UHD_ASSERT_THROW(num_bytes <= boost::asio::buffer_size(_buff));
-
-        _lut->length = num_bytes;
-        _lut->actual_length = 0;
-
+    void commit(libusb_transfer *lut, size_t num_bytes){
+        lut->length = num_bytes;
         try{
-            _endpoint->submit(_lut);
-            _committed = true;
-            return num_bytes;
+            _send_ep->submit(lut);
         }
         catch(const std::exception &e){
             std::cerr << "Error in commit: " << e.what() << std::endl;
-            return -1;
         }
     }
 
-private:
-    const boost::asio::mutable_buffer &get(void) const{
-        return _buff;
-    }
-
-    libusb_transfer *_lut;
-    usb_endpoint::sptr _endpoint;
-    const boost::asio::mutable_buffer _buff;
-    bool _committed;
-};
-
-
-/***********************************************************************
- * USB zero_copy device class
- **********************************************************************/
-class libusb_zero_copy_impl : public usb_zero_copy
-{
-private:
     libusb::device_handle::sptr _handle;
+    size_t _recv_xfer_size, _send_xfer_size;
     size_t _recv_num_frames, _send_num_frames;
     usb_endpoint::sptr _recv_ep, _send_ep;
-
-public:
-    typedef boost::shared_ptr<libusb_zero_copy_impl> sptr;
-
-    libusb_zero_copy_impl(
-        libusb::device_handle::sptr handle,
-        unsigned int recv_endpoint, unsigned int send_endpoint,
-        size_t recv_xfer_size, size_t recv_num_xfers,
-        size_t send_xfer_size, size_t send_num_xfers
-    );
-
-    managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms);
-    managed_send_buffer::sptr get_send_buff(void);
-
-    size_t get_num_recv_frames(void) const { return _recv_num_frames; }
-    size_t get_num_send_frames(void) const { return _send_num_frames; }
 };
 
 /*
@@ -430,7 +340,9 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(
     UHD_ASSERT_THROW(send_xfer_size % 512 == 0);
 
     //store the num xfers for the num frames count
+    _recv_xfer_size = recv_xfer_size;
     _recv_num_frames = recv_num_xfers;
+    _send_xfer_size = send_xfer_size;
     _send_num_frames = send_num_xfers;
 
     _handle->claim_interface(2 /*in interface*/);
@@ -459,15 +371,16 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(
  * Return empty pointer if no transfer is available (timeout or error).
  * \return pointer to a managed receive buffer
  */
-managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(size_t timeout_ms){
-    libusb_transfer *lut = _recv_ep->get_lut_with_wait(timeout_ms);
+managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(double timeout){
+    libusb_transfer *lut = _recv_ep->get_lut_with_wait(timeout);
     if (lut == NULL) {
         return managed_recv_buffer::sptr();
     }
     else {
-        return managed_recv_buffer::sptr(
-            new libusb_managed_recv_buffer_impl(lut,
-                                                _recv_ep));
+        return managed_recv_buffer::make_safe(
+            boost::asio::const_buffer(lut->buffer, lut->actual_length),
+            boost::bind(&libusb_zero_copy_impl::release, shared_from_this(), lut)
+        );
     }
 }
 
@@ -478,15 +391,16 @@ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(size_t timeout_ms
  * (timeout or error).
  * \return pointer to a managed send buffer
  */
-managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(void){
-    libusb_transfer *lut = _send_ep->get_lut_with_wait(/* TODO timeout API */);
+managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){
+    libusb_transfer *lut = _send_ep->get_lut_with_wait(timeout);
     if (lut == NULL) {
         return managed_send_buffer::sptr();
     }
     else {
-        return managed_send_buffer::sptr(
-            new libusb_managed_send_buffer_impl(lut,
-                                                _send_ep));
+        return managed_send_buffer::make_safe(
+            boost::asio::mutable_buffer(lut->buffer, _send_xfer_size),
+            boost::bind(&libusb_zero_copy_impl::commit, shared_from_this(), lut, _1)
+        );
     }
 }
 
@@ -494,18 +408,18 @@ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(void){
  * USB zero_copy make functions
  **********************************************************************/
 usb_zero_copy::sptr usb_zero_copy::make(
-	usb_device_handle::sptr handle,
+    usb_device_handle::sptr handle,
     unsigned int recv_endpoint, unsigned int send_endpoint,
-	size_t recv_xfer_size, size_t recv_num_xfers,
-	size_t send_xfer_size, size_t send_num_xfers
+    size_t recv_xfer_size, size_t recv_num_xfers,
+    size_t send_xfer_size, size_t send_num_xfers
 ){
     libusb::device_handle::sptr dev_handle(libusb::device_handle::get_cached_handle(
         boost::static_pointer_cast<libusb::special_handle>(handle)->get_device()
     ));
     return sptr(new libusb_zero_copy_impl(
-		dev_handle,
-		recv_endpoint,  send_endpoint,
-		recv_xfer_size, recv_num_xfers,
-		send_xfer_size, send_num_xfers
+        dev_handle,
+        recv_endpoint,  send_endpoint,
+        recv_xfer_size, recv_num_xfers,
+        send_xfer_size, send_num_xfers
     ));
 }