uhd: implemented recv timeout for zero copy interface

6 files changed, 39 insertions, 36 deletions

diff --git a/host/include/uhd/transport/zero_copy.hpp b/host/include/uhd/transport/zero_copy.hpp
index 513291b63..8ecafd3fb 100644
--- a/host/include/uhd/transport/zero_copy.hpp
+++ b/host/include/uhd/transport/zero_copy.hpp
@@ -122,9 +122,10 @@ namespace uhd{ namespace transport{
 
         /*!
          * Get a new receive buffer from this transport object.
+         * \param timeout_ms the timeout to get the buffer in ms
          * \return a managed buffer, or null sptr on timeout/error
          */
-        virtual managed_recv_buffer::sptr get_recv_buff(void) = 0;
+        virtual managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms) = 0;
 
         /*!
          * Get the maximum number of receive frames:
@@ -171,16 +172,19 @@ namespace uhd{ namespace transport{
 
         /*!
          * Get a new receive buffer from this transport object.
+         * \param timeout_ms the timeout to get the buffer in ms
+         * \return a managed buffer, or null sptr on timeout/error
          */
-        managed_recv_buffer::sptr get_recv_buff(void);
+        managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms);
 
     private:
         /*!
          * Perform a private copying recv.
          * \param buff the buffer to write data into
+         * \param timeout_ms the timeout to get the buffer in ms
          * \return the number of bytes written to buff, 0 for timeout, negative for error
          */
-        virtual ssize_t recv(const boost::asio::mutable_buffer &buff) = 0;
+        virtual ssize_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms) = 0;
 
         UHD_PIMPL_DECL(impl) _impl;
     };
@@ -204,6 +208,7 @@ namespace uhd{ namespace transport{
 
         /*!
          * Get a new send buffer from this transport object.
+         * \return a managed buffer, or null sptr on timeout/error
          */
         managed_send_buffer::sptr get_send_buff(void);
 
diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index f2dcff6b5..f9beb0b4c 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -308,8 +308,7 @@ public:
     }
 
 private:
-    const boost::asio::const_buffer &get() const
-    {
+    const boost::asio::const_buffer &get(void) const{
         return _buff;
     }
 
@@ -369,8 +368,7 @@ public:
     }
 
 private:
-    const boost::asio::mutable_buffer &get() const
-    {
+    const boost::asio::mutable_buffer &get(void) const{
         return _buff;
     }
 
@@ -395,13 +393,13 @@ 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
-	);
+        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(void);
+    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; }
@@ -419,23 +417,23 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(
     size_t recv_xfer_size, size_t recv_num_xfers,
     size_t send_xfer_size, size_t send_num_xfers
 ){
-	_handle = handle;
+    _handle = handle;
 
-	//if the sizes are left at 0 (automatic) -> use the defaults
-	if (recv_xfer_size == 0) recv_xfer_size = DEFAULT_XFER_SIZE;
-	if (recv_num_xfers == 0) recv_num_xfers = DEFAULT_NUM_XFERS;
-	if (send_xfer_size == 0) send_xfer_size = DEFAULT_XFER_SIZE;
-	if (send_num_xfers == 0) send_num_xfers = DEFAULT_NUM_XFERS;
+    //if the sizes are left at 0 (automatic) -> use the defaults
+    if (recv_xfer_size == 0) recv_xfer_size = DEFAULT_XFER_SIZE;
+    if (recv_num_xfers == 0) recv_num_xfers = DEFAULT_NUM_XFERS;
+    if (send_xfer_size == 0) send_xfer_size = DEFAULT_XFER_SIZE;
+    if (send_num_xfers == 0) send_num_xfers = DEFAULT_NUM_XFERS;
 
     //sanity check the transfer sizes
     UHD_ASSERT_THROW(recv_xfer_size % 512 == 0);
     UHD_ASSERT_THROW(send_xfer_size % 512 == 0);
 
-	//store the num xfers for the num frames count
-	_recv_num_frames = recv_num_xfers;
-	_send_num_frames = send_num_xfers;
+    //store the num xfers for the num frames count
+    _recv_num_frames = recv_num_xfers;
+    _send_num_frames = send_num_xfers;
 
-	_handle->claim_interface(2 /*in interface*/);
+    _handle->claim_interface(2 /*in interface*/);
     _handle->claim_interface(1 /*out interface*/);
 
     _recv_ep = usb_endpoint::sptr(new usb_endpoint(
@@ -461,8 +459,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(void){
-    libusb_transfer *lut = _recv_ep->get_lut_with_wait(/* TODO timeout API */);
+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);
     if (lut == NULL) {
         return managed_recv_buffer::sptr();
     }
diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp
index ee989ee2b..0a6c9f2af 100644
--- a/host/lib/transport/udp_zero_copy_asio.cpp
+++ b/host/lib/transport/udp_zero_copy_asio.cpp
@@ -35,7 +35,6 @@ static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(sizeof(boost::uint32_t) * 2
 //Perhaps this is due to the kernel scheduling,
 //but may change with host-based flow control.
 static const size_t MIN_SEND_SOCK_BUFF_SIZE = size_t(10e3);
-static const double RECV_TIMEOUT = 0.1; //100 ms
 
 /***********************************************************************
  * Zero Copy UDP implementation with ASIO:
@@ -110,11 +109,11 @@ private:
     boost::asio::io_service        _io_service;
     int                            _sock_fd;
 
-    ssize_t recv(const boost::asio::mutable_buffer &buff){
+    ssize_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){
         //setup timeval for timeout
         timeval tv;
         tv.tv_sec = 0;
-        tv.tv_usec = int(RECV_TIMEOUT*1e6);
+        tv.tv_usec = timeout_ms*1000;
 
         //setup rset for timeout
         fd_set rset;
diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp
index 8a1cde694..1fcf846a0 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(void){
+managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(size_t timeout_ms){
     //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));
+    ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout_ms);
 
     if (num_bytes <= 0) return managed_recv_buffer::sptr(); //NULL sptr
 
diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp
index 73974f2d6..aee760a83 100644
--- a/host/lib/usrp/usrp1/io_impl.cpp
+++ b/host/lib/usrp/usrp1/io_impl.cpp
@@ -272,18 +272,18 @@ static void usrp1_bs_vrt_unpacker(
 }
 
 static bool get_recv_buffs(
-    zero_copy_if::sptr zc_if,
+    zero_copy_if::sptr zc_if, size_t timeout_ms,
     vrt_packet_handler::managed_recv_buffs_t &buffs
 ){
     UHD_ASSERT_THROW(buffs.size() == 1);
-    buffs[0] = zc_if->get_recv_buff();
+    buffs[0] = zc_if->get_recv_buff(timeout_ms);
     return buffs[0].get() != NULL;
 }
 
 size_t usrp1_impl::recv(
     const std::vector<void *> &buffs, size_t num_samps,
     rx_metadata_t &metadata, const io_type_t &io_type,
-    recv_mode_t recv_mode, size_t /*timeout_ms TODO*/
+    recv_mode_t recv_mode, size_t timeout_ms
 ){
     size_t num_samps_recvd = vrt_packet_handler::recv(
         _io_impl->packet_handler_recv_state,       //last state of the recv handler
@@ -292,7 +292,7 @@ size_t usrp1_impl::recv(
         io_type, _rx_otw_type,                     //input and output types to convert
         _clock_ctrl->get_master_clock_freq(),      //master clock tick rate
         &usrp1_bs_vrt_unpacker,
-        boost::bind(&get_recv_buffs, _data_transport, _1),
+        boost::bind(&get_recv_buffs, _data_transport, timeout_ms, _1),
         &vrt_packet_handler::handle_overflow_nop,
         0,                                         //vrt header offset
         _rx_subdev_spec.size()                     //num channels
diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp
index 65411801d..3395f94e2 100644
--- a/host/lib/usrp/usrp2/io_impl.cpp
+++ b/host/lib/usrp/usrp2/io_impl.cpp
@@ -33,6 +33,7 @@ using namespace uhd::transport;
 namespace asio = boost::asio;
 
 static const int underflow_flags = async_metadata_t::EVENT_CODE_UNDERFLOW | async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET;
+static const double RECV_TIMEOUT_MS = 100;
 
 /***********************************************************************
  * io impl details (internal to this file)
@@ -90,7 +91,7 @@ void usrp2_impl::io_impl::recv_pirate_loop(
     size_t next_packet_seq = 0;
 
     while(recv_pirate_crew_raiding){
-        managed_recv_buffer::sptr buff = zc_if->get_recv_buff();
+        managed_recv_buffer::sptr buff = zc_if->get_recv_buff(RECV_TIMEOUT_MS);
         if (not buff.get()) continue; //ignore timeout/error buffers
 
         try{
@@ -150,7 +151,7 @@ void usrp2_impl::io_init(void){
         std::memcpy(send_buff->cast<void*>(), &data, sizeof(data));
         send_buff->commit(sizeof(data));
         //drain the recv buffers (may have junk)
-        while (data_transport->get_recv_buff().get()){};
+        while (data_transport->get_recv_buff(RECV_TIMEOUT_MS).get()){};
     }
 
     //the number of recv frames is the number for the first transport