Merge branch 'udp_ports' into next

Conflicts:
	firmware/microblaze/apps/txrx_uhd.c
	host/lib/usrp/usrp2/mboard_impl.cpp
	host/lib/usrp/usrp2/usrp2_impl.cpp
	host/lib/usrp/usrp2/usrp2_impl.hpp

5 files changed, 196 insertions, 52 deletions

diff --git a/host/lib/usrp/usrp2/fw_common.h b/host/lib/usrp/usrp2/fw_common.h
index efbb4b954..ee7fc3882 100644
--- a/host/lib/usrp/usrp2/fw_common.h
+++ b/host/lib/usrp/usrp2/fw_common.h
@@ -42,7 +42,9 @@ extern "C" {
 // udp ports for the usrp2 communication
 // Dynamic and/or private ports: 49152-65535
 #define USRP2_UDP_CTRL_PORT 49152
-#define USRP2_UDP_DATA_PORT 49153
+//#define USRP2_UDP_UPDATE_PORT 49154
+#define USRP2_UDP_DATA_PORT 49156
+#define USRP2_UDP_ERR0_PORT 49157
 
 ////////////////////////////////////////////////////////////////////////
 // I2C addresses
diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp
index cbc0a0817..5a6c0983c 100644
--- a/host/lib/usrp/usrp2/io_impl.cpp
+++ b/host/lib/usrp/usrp2/io_impl.cpp
@@ -21,11 +21,13 @@
 #include <uhd/utils/byteswap.hpp>
 #include <uhd/utils/thread_priority.hpp>
 #include <uhd/transport/convert_types.hpp>
-#include <uhd/transport/alignment_buffer.hpp>
+#include <uhd/transport/bounded_buffer.hpp>
 #include <boost/format.hpp>
 #include <boost/bind.hpp>
 #include <boost/thread.hpp>
+#include <boost/date_time/posix_time/posix_time_types.hpp>
 #include <iostream>
+#include <list>
 
 using namespace uhd;
 using namespace uhd::usrp;
@@ -108,16 +110,24 @@ private:
  * - vrt packet handler states
  **********************************************************************/
 struct usrp2_impl::io_impl{
-    typedef alignment_buffer<managed_recv_buffer::sptr, time_spec_t> alignment_buffer_type;
 
-    io_impl(size_t num_recv_frames, size_t send_frame_size, size_t width):
+    io_impl(size_t send_frame_size, size_t width):
         packet_handler_recv_state(width),
-        recv_pirate_booty(alignment_buffer_type::make(num_recv_frames-3, width)),
         async_msg_fifo(bounded_buffer<async_metadata_t>::make(100/*messages deep*/))
     {
-        for (size_t i = 0; i < width; i++) fc_mons.push_back(
-            flow_control_monitor::sptr(new flow_control_monitor(usrp2_impl::sram_bytes/send_frame_size))
-        );
+        for (size_t i = 0; i < width; i++){
+            fc_mons.push_back(flow_control_monitor::sptr(
+                new flow_control_monitor(usrp2_impl::sram_bytes/send_frame_size)
+            ));
+            //init empty packet infos
+            vrt::if_packet_info_t packet_info;
+            packet_info.packet_count = 0xf;
+            packet_info.has_tsi = true;
+            packet_info.tsi = 0;
+            packet_info.has_tsf = true;
+            packet_info.tsf = 0;
+            prev_infos.push_back(packet_info);
+        }
     }
 
     ~io_impl(void){
@@ -126,11 +136,6 @@ struct usrp2_impl::io_impl{
         recv_pirate_crew.join_all();
     }
 
-    bool get_recv_buffs(vrt_packet_handler::managed_recv_buffs_t &buffs, double timeout){
-        boost::this_thread::disable_interruption di; //disable because the wait can throw
-        return recv_pirate_booty->pop_elems_with_timed_wait(buffs, timeout);
-    }
-
     bool get_send_buffs(
         const std::vector<zero_copy_if::sptr> &trans,
         vrt_packet_handler::managed_send_buffs_t &buffs,
@@ -151,6 +156,15 @@ struct usrp2_impl::io_impl{
         return true;
     }
 
+    bool get_recv_buffs(
+        const std::vector<zero_copy_if::sptr> &xports,
+        vrt_packet_handler::managed_recv_buffs_t &buffs,
+        double timeout
+    );
+
+    //previous state for each buffer
+    std::vector<vrt::if_packet_info_t> prev_infos;
+
     //flow control monitors
     std::vector<flow_control_monitor::sptr> fc_mons;
 
@@ -162,29 +176,28 @@ struct usrp2_impl::io_impl{
     void recv_pirate_loop(zero_copy_if::sptr, usrp2_mboard_impl::sptr, size_t);
     boost::thread_group recv_pirate_crew;
     bool recv_pirate_crew_raiding;
-    alignment_buffer_type::sptr recv_pirate_booty;
     bounded_buffer<async_metadata_t>::sptr async_msg_fifo;
     boost::mutex spawn_mutex;
 };
 
 /***********************************************************************
  * Receive Pirate Loop
- * - while raiding, loot for recv buffers
- * - put booty into the alignment buffer
+ * - while raiding, loot for message packet
+ * - update flow control condition count
+ * - put async message packets into queue
  **********************************************************************/
 void usrp2_impl::io_impl::recv_pirate_loop(
-    zero_copy_if::sptr zc_if,
+    zero_copy_if::sptr zc_if_err0,
     usrp2_mboard_impl::sptr mboard,
     size_t index
 ){
     set_thread_priority_safe();
     recv_pirate_crew_raiding = true;
-    size_t next_packet_seq = 0;
 
     spawn_mutex.unlock();
 
     while(recv_pirate_crew_raiding){
-        managed_recv_buffer::sptr buff = zc_if->get_recv_buff();
+        managed_recv_buffer::sptr buff = zc_if_err0->get_recv_buff();
         if (not buff.get()) continue; //ignore timeout/error buffers
 
         try{
@@ -194,26 +207,6 @@ void usrp2_impl::io_impl::recv_pirate_loop(
             const boost::uint32_t *vrt_hdr = buff->cast<const boost::uint32_t *>();
             vrt::if_hdr_unpack_be(vrt_hdr, if_packet_info);
 
-            //handle the rx data stream
-            if (if_packet_info.sid == usrp2_impl::RECV_SID){
-                //handle the packet count / sequence number
-                if (if_packet_info.packet_count != next_packet_seq){
-                    //std::cerr << "S" << (if_packet_info.packet_count - next_packet_seq)%16;
-                    std::cerr << "O" << std::flush; //report overflow (drops in the kernel)
-                }
-                next_packet_seq = (if_packet_info.packet_count+1)%16;
-
-                //extract the timespec and round to the nearest packet
-                UHD_ASSERT_THROW(if_packet_info.has_tsi and if_packet_info.has_tsf);
-                time_spec_t time(
-                    time_t(if_packet_info.tsi), size_t(if_packet_info.tsf), mboard->get_master_clock_freq()
-                );
-
-                //push the packet into the buffer with the new time
-                recv_pirate_booty->push_with_pop_on_full(buff, time, index);
-                continue;
-            }
-
             //handle a tx async report message
             if (if_packet_info.sid == usrp2_impl::ASYNC_SID and if_packet_info.packet_type != vrt::if_packet_info_t::PACKET_TYPE_DATA){
 
@@ -253,11 +246,10 @@ void usrp2_impl::io_impl::recv_pirate_loop(
 void usrp2_impl::io_init(void){
 
     //the assumption is that all data transports should be identical
-    const size_t num_recv_frames = _data_transports.front()->get_num_recv_frames();
     const size_t send_frame_size = _data_transports.front()->get_send_frame_size();
 
     //create new io impl
-    _io_impl = UHD_PIMPL_MAKE(io_impl, (num_recv_frames, send_frame_size, _data_transports.size()));
+    _io_impl = UHD_PIMPL_MAKE(io_impl, (send_frame_size, _data_transports.size()));
 
     //TODO temporary fix for weird power up state, remove when FPGA fixed
     {
@@ -276,7 +268,7 @@ void usrp2_impl::io_init(void){
         //spawn a new pirate to plunder the recv booty
         _io_impl->recv_pirate_crew.create_thread(boost::bind(
             &usrp2_impl::io_impl::recv_pirate_loop,
-            _io_impl.get(), _data_transports.at(i),
+            _io_impl.get(), _err0_transports.at(i),
             _mboards.at(i), i
         ));
         //block here until the spawned thread unlocks
@@ -328,6 +320,133 @@ size_t usrp2_impl::send(
 }
 
 /***********************************************************************
+ * Alignment logic on receive
+ **********************************************************************/
+static UHD_INLINE boost::posix_time::time_duration to_time_dur(double timeout){
+    return boost::posix_time::microseconds(long(timeout*1e6));
+}
+
+static UHD_INLINE double from_time_dur(const boost::posix_time::time_duration &time_dur){
+    return 1e-6*time_dur.total_microseconds();
+}
+
+static UHD_INLINE time_spec_t extract_time_spec(
+    const vrt::if_packet_info_t &packet_info
+){
+    return time_spec_t( //assumes has_tsi and has_tsf are true
+        time_t(packet_info.tsi), size_t(packet_info.tsf),
+        100e6 //tick rate does not have to be correct for comparison purposes
+    );
+}
+
+static UHD_INLINE void extract_packet_info(
+    managed_recv_buffer::sptr &buff,
+    vrt::if_packet_info_t &prev_info,
+    time_spec_t &time, bool &clear, bool &msg
+){
+    //extract packet info
+    vrt::if_packet_info_t next_info;
+    next_info.num_packet_words32 = buff->size()/sizeof(boost::uint32_t);
+    vrt::if_hdr_unpack_be(buff->cast<const boost::uint32_t *>(), next_info);
+
+    //handle the packet count / sequence number
+    if ((prev_info.packet_count+1)%16 != next_info.packet_count){
+        std::cerr << "O" << std::flush; //report overflow (drops in the kernel)
+    }
+
+    time = extract_time_spec(next_info);
+    clear = extract_time_spec(prev_info) > time;
+    msg = next_info.packet_type != vrt::if_packet_info_t::PACKET_TYPE_DATA;
+    prev_info = next_info;
+}
+
+static UHD_INLINE bool handle_msg_packet(
+    vrt_packet_handler::managed_recv_buffs_t &buffs, size_t index
+){
+    for (size_t i = 0; i < buffs.size(); i++){
+        if (i == index) continue;
+        buffs[i].reset(); //set NULL
+    }
+    return true;
+}
+
+UHD_INLINE bool usrp2_impl::io_impl::get_recv_buffs(
+    const std::vector<zero_copy_if::sptr> &xports,
+    vrt_packet_handler::managed_recv_buffs_t &buffs,
+    double timeout
+){
+    if (buffs.size() == 1){
+        buffs[0] = xports[0]->get_recv_buff(timeout);
+        if (buffs[0].get() == NULL) return false;
+        bool clear, msg; time_spec_t time; //unused variables
+        //call extract_packet_info to handle printing the overflows
+        extract_packet_info(buffs[0], this->prev_infos[0], time, clear, msg);
+        return true;
+    }
+    //-------------------- begin alignment logic ---------------------//
+    boost::system_time exit_time = boost::get_system_time() + to_time_dur(timeout);
+    managed_recv_buffer::sptr buff_tmp;
+    std::list<size_t> _all_indexes, indexes_to_do;
+    for (size_t i = 0; i < buffs.size(); i++) _all_indexes.push_back(i);
+    bool clear, msg;
+    time_spec_t expected_time;
+
+    //respond to a clear by starting from scratch
+    got_clear:
+    indexes_to_do = _all_indexes;
+    clear = false;
+
+    //do an initial pop to load an initial sequence id
+    size_t index = indexes_to_do.front();
+    buff_tmp = xports[index]->get_recv_buff(from_time_dur(exit_time - boost::get_system_time()));
+    if (buff_tmp.get() == NULL) return false;
+    extract_packet_info(buff_tmp, this->prev_infos[index], expected_time, clear, msg);
+    if (clear) goto got_clear;
+    buffs[index] = buff_tmp;
+    if (msg) return handle_msg_packet(buffs, index);
+    indexes_to_do.pop_front();
+
+    //get an aligned set of elements from the buffers:
+    while(indexes_to_do.size() != 0){
+
+        //pop an element off for this index
+        index = indexes_to_do.front();
+        buff_tmp = xports[index]->get_recv_buff(from_time_dur(exit_time - boost::get_system_time()));
+        if (buff_tmp.get() == NULL) return false;
+        time_spec_t this_time;
+        extract_packet_info(buff_tmp, this->prev_infos[index], this_time, clear, msg);
+        if (clear) goto got_clear;
+        buffs[index] = buff_tmp;
+        if (msg) return handle_msg_packet(buffs, index);
+
+        //if the sequence id matches:
+        //  remove this index from the list and continue
+        if (this_time == expected_time){
+            indexes_to_do.pop_front();
+            continue;
+        }
+
+        //if the sequence id is older:
+        //  continue with the same index to try again
+        else if (this_time < expected_time){
+            continue;
+        }
+
+        //if the sequence id is newer:
+        //  use the new expected time for comparison
+        //  add all other indexes back into the list
+        else{
+            expected_time = this_time;
+            indexes_to_do = _all_indexes;
+            indexes_to_do.remove(index);
+            continue;
+        }
+    }
+    return true;
+    //-------------------- end alignment logic -----------------------//
+}
+
+/***********************************************************************
  * Receive Data
  **********************************************************************/
 size_t usrp2_impl::get_max_recv_samps_per_packet(void) const{
@@ -357,7 +476,7 @@ size_t usrp2_impl::recv(
         io_type, _rx_otw_type,                     //input and output types to convert
         _mboards.front()->get_master_clock_freq(), //master clock tick rate
         uhd::transport::vrt::if_hdr_unpack_be,
-        boost::bind(&usrp2_impl::io_impl::get_recv_buffs, _io_impl.get(), _1, timeout),
+        boost::bind(&usrp2_impl::io_impl::get_recv_buffs, _io_impl.get(), _data_transports, _1, timeout),
         boost::bind(&handle_overflow, _mboards, _1)
     );
 }
diff --git a/host/lib/usrp/usrp2/mboard_impl.cpp b/host/lib/usrp/usrp2/mboard_impl.cpp
index b8ebd6030..72d1c9d03 100644
--- a/host/lib/usrp/usrp2/mboard_impl.cpp
+++ b/host/lib/usrp/usrp2/mboard_impl.cpp
@@ -42,6 +42,7 @@ usrp2_mboard_impl::usrp2_mboard_impl(
     size_t index,
     transport::udp_simple::sptr ctrl_transport,
     transport::zero_copy_if::sptr data_transport,
+    transport::zero_copy_if::sptr err0_transport,
     const device_addr_t &device_args,
     size_t recv_samps_per_packet
 ):
@@ -51,11 +52,15 @@ usrp2_mboard_impl::usrp2_mboard_impl(
     //Send a small data packet so the usrp2 knows the udp source port.
     //This setup must happen before further initialization occurs
     //or the async update packets will cause ICMP destination unreachable.
-    transport::managed_send_buffer::sptr send_buff = data_transport->get_send_buff();
+    transport::managed_send_buffer::sptr send_buff;
     static const boost::uint32_t data[2] = {
         uhd::htonx(boost::uint32_t(0 /* don't care seq num */)),
         uhd::htonx(boost::uint32_t(USRP2_INVALID_VRT_HEADER))
     };
+    send_buff = data_transport->get_send_buff();
+    std::memcpy(send_buff->cast<void*>(), &data, sizeof(data));
+    send_buff->commit(sizeof(data));
+    send_buff = err0_transport->get_send_buff();
     std::memcpy(send_buff->cast<void*>(), &data, sizeof(data));
     send_buff->commit(sizeof(data));
 
diff --git a/host/lib/usrp/usrp2/usrp2_impl.cpp b/host/lib/usrp/usrp2/usrp2_impl.cpp
index 133c39a35..f910999d4 100644
--- a/host/lib/usrp/usrp2/usrp2_impl.cpp
+++ b/host/lib/usrp/usrp2/usrp2_impl.cpp
@@ -197,10 +197,18 @@ static device_addrs_t usrp2_find(const device_addr_t &hint_){
  * Make
  **********************************************************************/
 static device::sptr usrp2_make(const device_addr_t &device_addr){
-sep_indexed_dev_addrs(device_addr);
+
+    //setup the dsp transport hints (default to a large recv buff)
+    device_addr_t dsp_xport_hints = device_addr;
+    if (not dsp_xport_hints.has_key("recv_buff_size")){
+        //set to half-a-second of buffering at max rate
+        dsp_xport_hints["recv_buff_size"] = "50e6";
+    }
+
     //create a ctrl and data transport for each address
     std::vector<udp_simple::sptr> ctrl_transports;
     std::vector<zero_copy_if::sptr> data_transports;
+    std::vector<zero_copy_if::sptr> err0_transports;
     const device_addrs_t device_addrs = sep_indexed_dev_addrs(device_addr);
 
     BOOST_FOREACH(const device_addr_t &dev_addr_i, device_addrs){
@@ -208,14 +216,17 @@ sep_indexed_dev_addrs(device_addr);
             dev_addr_i["addr"], num2str(USRP2_UDP_CTRL_PORT)
         ));
         data_transports.push_back(udp_zero_copy::make(
-            dev_addr_i["addr"], num2str(USRP2_UDP_DATA_PORT), device_addr
+            dev_addr_i["addr"], num2str(USRP2_UDP_DATA_PORT), dsp_xport_hints
+        ));
+        err0_transports.push_back(udp_zero_copy::make(
+            dev_addr_i["addr"], num2str(USRP2_UDP_ERR0_PORT), device_addr_t()
         ));
     }
 
     //create the usrp2 implementation guts
-    return device::sptr(
-        new usrp2_impl(ctrl_transports, data_transports, device_addrs)
-    );
+    return device::sptr(new usrp2_impl(
+        ctrl_transports, data_transports, err0_transports, device_addrs
+    ));
 }
 
 UHD_STATIC_BLOCK(register_usrp2_device){
@@ -228,9 +239,11 @@ UHD_STATIC_BLOCK(register_usrp2_device){
 usrp2_impl::usrp2_impl(
     std::vector<udp_simple::sptr> ctrl_transports,
     std::vector<zero_copy_if::sptr> data_transports,
+    std::vector<zero_copy_if::sptr> err0_transports,
     const device_addrs_t &device_args
 ):
-    _data_transports(data_transports)
+    _data_transports(data_transports),
+    _err0_transports(err0_transports)
 {
     //setup rx otw type
     _rx_otw_type.width = 16;
@@ -247,7 +260,8 @@ usrp2_impl::usrp2_impl(
     //create a new mboard handler for each control transport
     for(size_t i = 0; i < device_args.size(); i++){
         _mboards.push_back(usrp2_mboard_impl::sptr(new usrp2_mboard_impl(
-            i, ctrl_transports[i], data_transports[i], device_args[i],
+            i, ctrl_transports[i], data_transports[i],
+            err0_transports[i], device_args[i],
             this->get_max_recv_samps_per_packet()
         )));
         //use an empty name when there is only one mboard
diff --git a/host/lib/usrp/usrp2/usrp2_impl.hpp b/host/lib/usrp/usrp2/usrp2_impl.hpp
index 85c00b079..9cd27ee41 100644
--- a/host/lib/usrp/usrp2/usrp2_impl.hpp
+++ b/host/lib/usrp/usrp2/usrp2_impl.hpp
@@ -85,6 +85,7 @@ public:
         size_t index,
         uhd::transport::udp_simple::sptr,
         uhd::transport::zero_copy_if::sptr,
+        uhd::transport::zero_copy_if::sptr,
         const uhd::device_addr_t &device_args,
         size_t recv_samps_per_packet
     );
@@ -186,11 +187,13 @@ public:
      * Create a new usrp2 impl base.
      * \param ctrl_transports the udp transports for control
      * \param data_transports the udp transports for data
-     * \param flow_control_hints optional flow control params
+     * \param err0_transports the udp transports for error
+     * \param device_args optional misc device parameters
      */
     usrp2_impl(
         std::vector<uhd::transport::udp_simple::sptr> ctrl_transports,
         std::vector<uhd::transport::zero_copy_if::sptr> data_transports,
+        std::vector<uhd::transport::zero_copy_if::sptr> err0_transports,
         const uhd::device_addrs_t &device_args
     );
 
@@ -222,6 +225,7 @@ private:
 
     //io impl methods and members
     std::vector<uhd::transport::zero_copy_if::sptr> _data_transports;
+    std::vector<uhd::transport::zero_copy_if::sptr> _err0_transports;
     uhd::otw_type_t _rx_otw_type, _tx_otw_type;
     UHD_PIMPL_DECL(io_impl) _io_impl;
     void io_init(void);