uhd: implemented a double timeout (in seconds) for send and recv chains

converted all size_t timeout_ms to double timeout

bounded and alignment buffer now take double timeout

added timeout to device::send and zero_copy_if::get_send_buff

4 files changed, 50 insertions, 51 deletions

diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index f9beb0b4c..7f2bc3468 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);
@@ -187,7 +185,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){
@@ -274,12 +272,10 @@ 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;
 }
 
@@ -399,8 +395,8 @@ public:
         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);
+    managed_recv_buffer::sptr get_recv_buff(double);
+    managed_send_buffer::sptr get_send_buff(double);
 
     size_t get_num_recv_frames(void) const { return _recv_num_frames; }
     size_t get_num_send_frames(void) const { return _send_num_frames; }
@@ -459,8 +455,8 @@ 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();
     }
@@ -478,8 +474,8 @@ 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();
     }
@@ -494,18 +490,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
     ));
 }
diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp
index 0a6c9f2af..3130830a5 100644
--- a/host/lib/transport/udp_zero_copy_asio.cpp
+++ b/host/lib/transport/udp_zero_copy_asio.cpp
@@ -109,11 +109,11 @@ private:
     boost::asio::io_service        _io_service;
     int                            _sock_fd;
 
-    ssize_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){
+    ssize_t recv(const boost::asio::mutable_buffer &buff, double timeout){
         //setup timeval for timeout
         timeval tv;
         tv.tv_sec = 0;
-        tv.tv_usec = timeout_ms*1000;
+        tv.tv_usec = long(timeout*1e6);
 
         //setup rset for timeout
         fd_set rset;
diff --git a/host/lib/transport/vrt_packet_handler.hpp b/host/lib/transport/vrt_packet_handler.hpp
index b603f1371..e11afff30 100644
--- a/host/lib/transport/vrt_packet_handler.hpp
+++ b/host/lib/transport/vrt_packet_handler.hpp
@@ -303,18 +303,18 @@ template <typename T> UHD_INLINE T get_context_code(
      * Pack a vrt header, copy-convert the data, and send it.
      *  - helper function for vrt_packet_handler::send
      ******************************************************************/
-    static UHD_INLINE void _send1(
+    static UHD_INLINE size_t _send1(
         send_state &state,
         const std::vector<const void *> &buffs,
-        size_t offset_bytes,
-        size_t num_samps,
+        const size_t offset_bytes,
+        const size_t num_samps,
         uhd::transport::vrt::if_packet_info_t &if_packet_info,
         const uhd::io_type_t &io_type,
         const uhd::otw_type_t &otw_type,
         const vrt_packer_t &vrt_packer,
         const get_send_buffs_t &get_send_buffs,
-        size_t vrt_header_offset_words32,
-        size_t chans_per_otw_buff
+        const size_t vrt_header_offset_words32,
+        const size_t chans_per_otw_buff
     ){
         //load the rest of the if_packet_info in here
         if_packet_info.num_payload_words32 = (num_samps*chans_per_otw_buff*otw_type.get_sample_size())/sizeof(boost::uint32_t);
@@ -322,7 +322,7 @@ template <typename T> UHD_INLINE T get_context_code(
 
         //get send buffers for each channel
         managed_send_buffs_t send_buffs(buffs.size()/chans_per_otw_buff);
-        UHD_ASSERT_THROW(get_send_buffs(send_buffs));
+        if (not get_send_buffs(send_buffs)) return 0;
 
         std::vector<const void *> io_buffs(chans_per_otw_buff);
         for (size_t i = 0; i < buffs.size(); i+=chans_per_otw_buff){
@@ -347,6 +347,7 @@ template <typename T> UHD_INLINE T get_context_code(
                 std::cerr << "commit to send buffer returned less than commit size" << std::endl;
             }
         }
+        return num_samps;
     }
 
     /*******************************************************************
@@ -381,7 +382,6 @@ template <typename T> UHD_INLINE T get_context_code(
         ////////////////////////////////////////////////////////////////
         case uhd::device::SEND_MODE_ONE_PACKET:{
         ////////////////////////////////////////////////////////////////
-            size_t num_samps = std::min(total_num_samps, max_samples_per_packet);
 
             //fill in parts of the packet info overwrote in full buff mode
             if_packet_info.has_tsi = metadata.has_time_spec;
@@ -389,10 +389,10 @@ template <typename T> UHD_INLINE T get_context_code(
             if_packet_info.sob = metadata.start_of_burst;
             if_packet_info.eob = metadata.end_of_burst;
 
-            _send1(
+            return _send1(
                 state,
                 buffs, 0,
-                num_samps,
+                std::min(total_num_samps, max_samples_per_packet),
                 if_packet_info,
                 io_type, otw_type,
                 vrt_packer,
@@ -400,31 +400,32 @@ template <typename T> UHD_INLINE T get_context_code(
                 vrt_header_offset_words32,
                 chans_per_otw_buff
             );
-            return num_samps;
         }
 
         ////////////////////////////////////////////////////////////////
         case uhd::device::SEND_MODE_FULL_BUFF:{
         ////////////////////////////////////////////////////////////////
-            //calculate constants for fragmentation
-            const size_t num_fragments = (total_num_samps+max_samples_per_packet-1)/max_samples_per_packet;
-            static const size_t first_fragment_index = 0;
-            const size_t final_fragment_index = num_fragments-1;
+            size_t total_num_samps_sent = 0;
 
             //loop through the following fragment indexes
-            for (size_t n = first_fragment_index; n <= final_fragment_index; n++){
+            while(total_num_samps_sent < total_num_samps){
+
+                //calculate per-loop-iteration variables
+                const size_t total_num_samps_unsent = total_num_samps - total_num_samps_sent;
+                const bool first_fragment = (total_num_samps_sent == 0);
+                const bool final_fragment = (total_num_samps_unsent <= max_samples_per_packet);
 
                 //calculate new flags for the fragments
-                if_packet_info.has_tsi = metadata.has_time_spec  and (n == first_fragment_index);
-                if_packet_info.has_tsf = metadata.has_time_spec  and (n == first_fragment_index);
-                if_packet_info.sob     = metadata.start_of_burst and (n == first_fragment_index);
-                if_packet_info.eob     = metadata.end_of_burst   and (n == final_fragment_index);
+                if_packet_info.has_tsi = metadata.has_time_spec  and first_fragment;
+                if_packet_info.has_tsf = if_packet_info.has_tsi;
+                if_packet_info.sob     = metadata.start_of_burst and first_fragment;
+                if_packet_info.eob     = metadata.end_of_burst   and final_fragment;
 
                 //send the fragment with the helper function
-                _send1(
+                const size_t num_samps_sent = _send1(
                     state,
-                    buffs, n*max_samples_per_packet*io_type.size,
-                    (n == final_fragment_index)?(total_num_samps%max_samples_per_packet):max_samples_per_packet,
+                    buffs, total_num_samps_sent*io_type.size,
+                    std::min(total_num_samps_unsent, max_samples_per_packet),
                     if_packet_info,
                     io_type, otw_type,
                     vrt_packer,
@@ -432,8 +433,10 @@ template <typename T> UHD_INLINE T get_context_code(
                     vrt_header_offset_words32,
                     chans_per_otw_buff
                 );
+                total_num_samps_sent += num_samps_sent;
+                if (num_samps_sent == 0) return total_num_samps_sent;
             }
-            return total_num_samps;
+            return total_num_samps_sent;
         }
 
         default: throw std::runtime_error("unknown send mode");
diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp
index 1fcf846a0..dfb65951f 100644
--- a/host/lib/transport/zero_copy.cpp
+++ b/host/lib/transport/zero_copy.cpp
@@ -68,12 +68,12 @@ phony_zero_copy_recv_if::~phony_zero_copy_recv_if(void){
     /* NOP */
 }
 
-managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(size_t timeout_ms){
+managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(double timeout){
     //allocate memory
     boost::uint8_t *recv_mem = new boost::uint8_t[_impl->max_buff_size];
 
     //call recv() with timeout option
-    ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout_ms);
+    ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout);
 
     if (num_bytes <= 0) return managed_recv_buffer::sptr(); //NULL sptr
 
@@ -138,6 +138,6 @@ phony_zero_copy_send_if::~phony_zero_copy_send_if(void){
     delete [] _impl->send_mem;
 }
 
-managed_send_buffer::sptr phony_zero_copy_send_if::get_send_buff(void){
+managed_send_buffer::sptr phony_zero_copy_send_if::get_send_buff(double){
     return _impl->send_buff; //FIXME there is only ever one send buff, we assume that the caller doesnt hang onto these
 }