Moved timeouts into the udp transports.

Simplified the fast path checking in the fw, but it turns out this was not the issue.
Fixed some bad bit operations with the 16sc words (dont forget sign extension).
Added some more documentation to the headers....

8 files changed, 104 insertions, 45 deletions

diff --git a/host/include/uhd/device.hpp b/host/include/uhd/device.hpp
index 1a52867c9..47dfa4328 100644
--- a/host/include/uhd/device.hpp
+++ b/host/include/uhd/device.hpp
@@ -77,6 +77,9 @@ public:
      * It is up to the caller to call send again on the un-sent
      * portions of the buffer, until the buffer is exhausted.
      *
+     * This is a blocking call and will not return until the number
+     * of samples returned have been read out of the buffer.
+     *
      * \param buff a buffer pointing to some read-only memory
      * \param metadata data describing the buffer's contents
      * \param the type of data loaded in the buffer (32fc, 16sc)
@@ -101,6 +104,14 @@ public:
      * The next call to receive, after the remainder becomes exahausted,
      * will perform an over-the-wire receive as usual.
      *
+     * This is a blocking call and will not return until the number
+     * of samples returned have been written into the buffer.
+     * However, a call to receive may timeout and return zero samples.
+     * The timeout duration is decided by the underlying transport layer.
+     * The caller should assume that the call to receive will not return
+     * immediately when no packets are available to the transport layer,
+     * and that the timeout duration is reasonably tuned for performance.
+     *
      * \param buff the buffer to fill with IF data
      * \param metadata data to fill describing the buffer
      * \param the type of data to fill into the buffer (32fc, 16sc)
diff --git a/host/include/uhd/transport/udp_simple.hpp b/host/include/uhd/transport/udp_simple.hpp
index 8663128ec..0d8fcc5f0 100644
--- a/host/include/uhd/transport/udp_simple.hpp
+++ b/host/include/uhd/transport/udp_simple.hpp
@@ -67,9 +67,9 @@ public:
 
     /*!
      * Receive into the provided buffer.
-     * Returns empty when data is not available.
+     * Blocks until data is received or a timeout occurs.
      * \param buff a mutable buffer to receive into
-     * \return the number of bytes received.
+     * \return the number of bytes received or zero on timeout
      */
     virtual size_t recv(const boost::asio::mutable_buffer &buff) = 0;
 };
diff --git a/host/include/uhd/transport/udp_zero_copy.hpp b/host/include/uhd/transport/udp_zero_copy.hpp
index 9c3505dd6..1a8d822fd 100644
--- a/host/include/uhd/transport/udp_zero_copy.hpp
+++ b/host/include/uhd/transport/udp_zero_copy.hpp
@@ -64,9 +64,9 @@ public:
 
     /*!
      * Receive a buffer.
+     * Blocks until data is received or a timeout occurs.
      * The memory is managed by the implementation.
-     * Returns an empty buffer when data is not available.
-     * \return a smart buffer with memory and size
+     * \return a smart buffer (empty on timeout)
      */
     virtual smart_buffer::sptr recv(void) = 0;
 };
diff --git a/host/lib/transport/udp_simple.cpp b/host/lib/transport/udp_simple.cpp
index 491cf59db..7004bdfdf 100644
--- a/host/lib/transport/udp_simple.cpp
+++ b/host/lib/transport/udp_simple.cpp
@@ -16,12 +16,35 @@
 //
 
 #include <uhd/transport/udp_simple.hpp>
+#include <boost/thread.hpp>
 #include <boost/format.hpp>
 #include <iostream>
 
 using namespace uhd::transport;
 
 /***********************************************************************
+ * Helper Functions
+ **********************************************************************/
+/*!
+ * A receive timeout for a socket:
+ *
+ * It seems that asio cannot have timeouts with synchronous io.
+ * However, we can implement a polling loop that will timeout.
+ * This is okay bacause this is the slow-path implementation.
+ *
+ * \param socket the asio socket
+ */
+static void reasonable_recv_timeout(
+    boost::asio::ip::udp::socket &socket
+){
+    boost::asio::deadline_timer timer(socket.get_io_service());
+    timer.expires_from_now(boost::posix_time::milliseconds(50));
+    while (not (socket.available() or timer.expires_from_now().is_negative())){
+        boost::this_thread::sleep(boost::posix_time::milliseconds(1));
+    }
+}
+
+/***********************************************************************
  * UDP connected implementation class
  **********************************************************************/
 class udp_connected_impl : public udp_simple{
@@ -62,7 +85,8 @@ size_t udp_connected_impl::send(const boost::asio::const_buffer &buff){
 }
 
 size_t udp_connected_impl::recv(const boost::asio::mutable_buffer &buff){
-    if (_socket->available() == 0) return 0;
+    reasonable_recv_timeout(*_socket);
+    if (not _socket->available()) return 0;
     return _socket->receive(boost::asio::buffer(buff));
 }
 
@@ -112,7 +136,8 @@ size_t udp_broadcast_impl::send(const boost::asio::const_buffer &buff){
 }
 
 size_t udp_broadcast_impl::recv(const boost::asio::mutable_buffer &buff){
-    if (_socket->available() == 0) return 0;
+    reasonable_recv_timeout(*_socket);
+    if (not _socket->available()) return 0;
     boost::asio::ip::udp::endpoint sender_endpoint;
     return _socket->receive_from(boost::asio::buffer(buff), sender_endpoint);
 }
diff --git a/host/lib/transport/udp_zero_copy_none.cpp b/host/lib/transport/udp_zero_copy_none.cpp
index e95706d94..e29530cf1 100644
--- a/host/lib/transport/udp_zero_copy_none.cpp
+++ b/host/lib/transport/udp_zero_copy_none.cpp
@@ -16,6 +16,8 @@
 //
 
 #include <uhd/transport/udp_zero_copy.hpp>
+#include <boost/thread.hpp>
+#include <boost/format.hpp>
 
 using namespace uhd::transport;
 
@@ -67,6 +69,9 @@ public:
 private:
     boost::asio::ip::udp::socket   *_socket;
     boost::asio::io_service        _io_service;
+
+    size_t get_recv_buff_size(void);
+    void set_recv_buff_size(size_t);
 };
 
 udp_zero_copy_impl::udp_zero_copy_impl(const std::string &addr, const std::string &port){
@@ -81,6 +86,18 @@ udp_zero_copy_impl::udp_zero_copy_impl(const std::string &addr, const std::strin
     _socket = new boost::asio::ip::udp::socket(_io_service);
     _socket->open(boost::asio::ip::udp::v4());
     _socket->connect(receiver_endpoint);
+
+    // set the rx socket buffer size:
+    // pick a huge size, and deal with whatever we get
+    set_recv_buff_size(54321e3); //some big number!
+    size_t current_buff_size = get_recv_buff_size();
+    std::cout << boost::format(
+        "Current rx socket buffer size: %d\n"
+    ) % current_buff_size;
+    if (current_buff_size < .1e6) std::cout << boost::format(
+        "Adjust max rx socket buffer size (linux only):\n"
+        "  sysctl -w net.core.rmem_max=VALUE\n"
+    );
 }
 
 udp_zero_copy_impl::~udp_zero_copy_impl(void){
@@ -92,14 +109,21 @@ size_t udp_zero_copy_impl::send(const boost::asio::const_buffer &buff){
 }
 
 smart_buffer::sptr udp_zero_copy_impl::recv(void){
-    size_t available = _socket->available();
+    size_t available = 0;
+
+    //implement timeout through polling and sleeping
+    boost::asio::deadline_timer timer(_socket->get_io_service());
+    timer.expires_from_now(boost::posix_time::milliseconds(50));
+    while (not ((available = _socket->available()) or timer.expires_from_now().is_negative())){
+        boost::this_thread::sleep(boost::posix_time::milliseconds(1));
+    }
 
     //allocate memory and create buffer
     uint32_t *buff_mem = new uint32_t[available/sizeof(uint32_t)];
     boost::asio::mutable_buffer buff(buff_mem, available);
 
     //receive only if data is available
-    if (available > 0){
+    if (available){
         _socket->receive(boost::asio::buffer(buff));
     }
 
@@ -107,6 +131,17 @@ smart_buffer::sptr udp_zero_copy_impl::recv(void){
     return smart_buffer::sptr(new smart_buffer_impl(buff));
 }
 
+size_t udp_zero_copy_impl::get_recv_buff_size(void){
+    boost::asio::socket_base::receive_buffer_size option;
+    _socket->get_option(option);
+    return option.value();
+}
+
+void udp_zero_copy_impl::set_recv_buff_size(size_t new_size){
+    boost::asio::socket_base::receive_buffer_size option(new_size);
+    _socket->set_option(option);
+}
+
 /***********************************************************************
  * UDP zero copy make function
  **********************************************************************/
diff --git a/host/lib/usrp/usrp2/dsp_impl.cpp b/host/lib/usrp/usrp2/dsp_impl.cpp
index a32f68872..7831b7667 100644
--- a/host/lib/usrp/usrp2/dsp_impl.cpp
+++ b/host/lib/usrp/usrp2/dsp_impl.cpp
@@ -21,6 +21,9 @@
 
 using namespace uhd;
 
+static const size_t default_decim = 16;
+static const size_t default_interp = 16;
+
 /***********************************************************************
  * DDC Helper Methods
  **********************************************************************/
@@ -37,7 +40,7 @@ static uint32_t calculate_freq_word_and_update_actual_freq(freq_t &freq, freq_t
 }
 
 static uint32_t calculate_iq_scale_word(int16_t i, int16_t q){
-    return ((i & 0xffff) << 16) | ((q & 0xffff) << 0);
+    return (uint16_t(i) << 16) | (uint16_t(q) << 0);
 }
 
 void usrp2_impl::init_ddc_config(void){
@@ -48,7 +51,7 @@ void usrp2_impl::init_ddc_config(void){
     );
 
     //initial config and update
-    _ddc_decim = 64;
+    _ddc_decim = default_decim;
     _ddc_freq = 0;
     update_ddc_config();
 
@@ -196,7 +199,7 @@ void usrp2_impl::init_duc_config(void){
     );
 
     //initial config and update
-    _duc_interp = 64;
+    _duc_interp = default_interp;
     _duc_freq = 0;
     update_duc_config();
 }
diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp
index 6273846dc..cc7746720 100644
--- a/host/lib/usrp/usrp2/io_impl.cpp
+++ b/host/lib/usrp/usrp2/io_impl.cpp
@@ -71,9 +71,9 @@ static inline void host_floats_to_usrp2_items(
     size_t num_samps
 ){
     unrolled_loop(i, num_samps,{
-        int16_t real = host_floats[i].real()*shorts_per_float;
-        int16_t imag = host_floats[i].imag()*shorts_per_float;
-        usrp2_items[i] = htonl(((real << 16) & 0xffff) | ((imag << 0) & 0xffff));
+        uint16_t real = host_floats[i].real()*shorts_per_float;
+        uint16_t imag = host_floats[i].imag()*shorts_per_float;
+        usrp2_items[i] = htonl((real << 16) | (imag << 0));
     });
 }
 
@@ -84,8 +84,8 @@ static inline void usrp2_items_to_host_floats(
 ){
     unrolled_loop(i, num_samps,{
         uint32_t item = ntohl(usrp2_items[i]);
-        int16_t real = (item >> 16) & 0xffff;
-        int16_t imag = (item >> 0)  & 0xffff;
+        int16_t real = item >> 16;
+        int16_t imag = item >> 0;
         host_floats[i] = fc32_t(real*floats_per_short, imag*floats_per_short);
     });
 }
@@ -130,9 +130,7 @@ void usrp2_impl::recv_raw(rx_metadata_t &metadata){
         return; //must exit here after setting the buffer
     }
     const uint32_t *vrt_hdr = asio::buffer_cast<const uint32_t *>(_rx_smart_buff->get());
-    size_t num_header_words32_out;
-    size_t num_payload_words32_out;
-    size_t packet_count_out;
+    size_t num_header_words32_out, num_payload_words32_out, packet_count_out;
     try{
         vrt::unpack(
             metadata,                //output
diff --git a/host/lib/usrp/usrp2/usrp2_impl.cpp b/host/lib/usrp/usrp2/usrp2_impl.cpp
index 752feb05b..58c82303f 100644
--- a/host/lib/usrp/usrp2/usrp2_impl.cpp
+++ b/host/lib/usrp/usrp2/usrp2_impl.cpp
@@ -24,6 +24,7 @@
 using namespace uhd;
 using namespace uhd::usrp;
 using namespace uhd::transport;
+namespace asio = boost::asio;
 
 /***********************************************************************
  * Discovery over the udp transport
@@ -43,19 +44,12 @@ uhd::device_addrs_t usrp2::discover(const device_addr_t &hint){
     ctrl_data_out.id = htonl(USRP2_CTRL_ID_GIVE_ME_YOUR_IP_ADDR_BRO);
     udp_transport->send(boost::asio::buffer(&ctrl_data_out, sizeof(ctrl_data_out)));
 
-    //loop and recieve until the time is up
-    size_t num_timeouts = 0;
+    //loop and recieve until the timeout
     while(true){
         usrp2_ctrl_data_t ctrl_data_in;
-        size_t len = udp_transport->recv(
-            boost::asio::buffer(&ctrl_data_in, sizeof(ctrl_data_in))
-        );
+        size_t len = udp_transport->recv(asio::buffer(&ctrl_data_in, sizeof(ctrl_data_in)));
         //std::cout << len << "\n";
-        if (len < sizeof(usrp2_ctrl_data_t)){
-            //sleep a little so we dont burn cpu
-            if (num_timeouts++ > 50) break;
-            boost::this_thread::sleep(boost::posix_time::milliseconds(1));
-        }else{
+        if (len >= sizeof(usrp2_ctrl_data_t)){
             //handle the received data
             switch(ntohl(ctrl_data_in.id)){
             case USRP2_CTRL_ID_THIS_IS_MY_IP_ADDR_DUDE:
@@ -67,9 +61,11 @@ uhd::device_addrs_t usrp2::discover(const device_addr_t &hint){
                 new_addr["transport"] = "udp";
                 new_addr["addr"] = ip_addr.to_string();
                 usrp2_addrs.push_back(new_addr);
-                break;
+                //dont break here, it will exit the while loop
+                //just continue on to the next loop iteration
             }
         }
+        if (len == 0) break; //timeout
     }
 
     return usrp2_addrs;
@@ -164,24 +160,15 @@ usrp2_ctrl_data_t usrp2_impl::ctrl_send_and_recv(const usrp2_ctrl_data_t &out_da
     out_copy.seq = htonl(++_ctrl_seq_num);
     _ctrl_transport->send(boost::asio::buffer(&out_copy, sizeof(usrp2_ctrl_data_t)));
 
-    //loop and recieve until the time is up
-    size_t num_timeouts = 0;
+    //loop until we get the packet or timeout
     while(true){
         usrp2_ctrl_data_t in_data;
-        size_t len = _ctrl_transport->recv(
-            boost::asio::buffer(&in_data, sizeof(in_data))
-        );
-        if (len < sizeof(usrp2_ctrl_data_t)){
-            //sleep a little so we dont burn cpu
-            if (num_timeouts++ > 50) break;
-            boost::this_thread::sleep(boost::posix_time::milliseconds(1));
-        }else{
-            //handle the received data
-            if (ntohl(in_data.seq) == _ctrl_seq_num){
-                return in_data;
-            }
-            //didnt get seq, continue on...
+        size_t len = _ctrl_transport->recv(asio::buffer(&in_data, sizeof(in_data)));
+        if (len >= sizeof(usrp2_ctrl_data_t) and ntohl(in_data.seq) == _ctrl_seq_num){
+            return in_data;
         }
+        if (len == 0) break; //timeout
+        //didnt get seq or bad packet, continue looking...
     }
     throw std::runtime_error("usrp2 no control response");
 }