Merge branch 'flow_ctrl_with_fpga'

10 files changed, 224 insertions, 108 deletions

diff --git a/host/lib/transport/udp_simple.cpp b/host/lib/transport/udp_simple.cpp
index 89750f99d..5829b462b 100644
--- a/host/lib/transport/udp_simple.cpp
+++ b/host/lib/transport/udp_simple.cpp
@@ -27,23 +27,25 @@ 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.
- *
+ * Wait for available data or timeout.
  * \param socket the asio socket
- * \param timeout_ms the timeout in milliseconds
+ * \param timeout the timeout in seconds
+ * \return false for timeout, true for data
  */
-static void reasonable_recv_timeout(
-    boost::asio::ip::udp::socket &socket, size_t timeout_ms
+static bool wait_available(
+    boost::asio::ip::udp::socket &socket, double timeout
 ){
-    boost::asio::deadline_timer timer(socket.get_io_service());
-    timer.expires_from_now(boost::posix_time::milliseconds(timeout_ms));
-    while (not (socket.available() or timer.expires_from_now().is_negative())){
-        boost::this_thread::sleep(boost::posix_time::milliseconds(1));
-    }
+    //setup timeval for timeout
+    timeval tv;
+    tv.tv_sec = 0;
+    tv.tv_usec = long(timeout*1e6);
+
+    //setup rset for timeout
+    fd_set rset;
+    FD_ZERO(&rset);
+    FD_SET(socket.native(), &rset);
+
+    return ::select(socket.native()+1, &rset, NULL, NULL, &tv) > 0;
 }
 
 /***********************************************************************
@@ -57,7 +59,7 @@ public:
 
     //send/recv
     size_t send(const boost::asio::const_buffer &);
-    size_t recv(const boost::asio::mutable_buffer &, size_t);
+    size_t recv(const boost::asio::mutable_buffer &, double);
 
 private:
     boost::asio::ip::udp::socket   *_socket;
@@ -86,9 +88,8 @@ size_t udp_connected_impl::send(const boost::asio::const_buffer &buff){
     return _socket->send(boost::asio::buffer(buff));
 }
 
-size_t udp_connected_impl::recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){
-    reasonable_recv_timeout(*_socket, timeout_ms);
-    if (not _socket->available()) return 0;
+size_t udp_connected_impl::recv(const boost::asio::mutable_buffer &buff, double timeout){
+    if (not wait_available(*_socket, timeout)) return 0;
     return _socket->receive(boost::asio::buffer(buff));
 }
 
@@ -103,7 +104,7 @@ public:
 
     //send/recv
     size_t send(const boost::asio::const_buffer &);
-    size_t recv(const boost::asio::mutable_buffer &, size_t);
+    size_t recv(const boost::asio::mutable_buffer &, double);
 
 private:
     boost::asio::ip::udp::socket   *_socket;
@@ -137,9 +138,8 @@ size_t udp_broadcast_impl::send(const boost::asio::const_buffer &buff){
     return _socket->send_to(boost::asio::buffer(buff), _receiver_endpoint);
 }
 
-size_t udp_broadcast_impl::recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){
-    reasonable_recv_timeout(*_socket, timeout_ms);
-    if (not _socket->available()) return 0;
+size_t udp_broadcast_impl::recv(const boost::asio::mutable_buffer &buff, double timeout){
+    if (not wait_available(*_socket, timeout)) 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_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp
index d84aeefdd..938ae4473 100644
--- a/host/lib/transport/udp_zero_copy_asio.cpp
+++ b/host/lib/transport/udp_zero_copy_asio.cpp
@@ -59,16 +59,23 @@ static const size_t DEFAULT_NUM_RECV_FRAMES = 32;
 #else
 static const size_t DEFAULT_NUM_RECV_FRAMES = MIN_RECV_SOCK_BUFF_SIZE/udp_simple::mtu;
 #endif
+
 //The non-async send only ever requires a single frame
 //because the buffer will be committed before a new get.
 #ifdef USE_ASIO_ASYNC_SEND
 static const size_t DEFAULT_NUM_SEND_FRAMES = 32;
 #else
-static const size_t DEFAULT_NUM_SEND_FRAMES = MIN_SEND_SOCK_BUFF_SIZE/udp_simple::mtu;;
+static const size_t DEFAULT_NUM_SEND_FRAMES = MIN_SEND_SOCK_BUFF_SIZE/udp_simple::mtu;
 #endif
 
-//a single concurrent thread for io_service seems to be the fastest
+//The number of service threads to spawn for async ASIO:
+//A single concurrent thread for io_service seems to be the fastest.
+//Threads are disabled when no async implementations are enabled.
+#if defined(USE_ASIO_ASYNC_RECV) || defined(USE_ASIO_ASYNC_SEND)
 static const size_t CONCURRENCY_HINT = 1;
+#else
+static const size_t CONCURRENCY_HINT = 0;
+#endif
 
 /***********************************************************************
  * Zero Copy UDP implementation with ASIO:
@@ -86,11 +93,12 @@ public:
         const std::string &port,
         const device_addr_t &hints
     ):
-        _io_service(hints.cast<size_t>("concurrency_hint", CONCURRENCY_HINT)),
         _recv_frame_size(size_t(hints.cast<double>("recv_frame_size", udp_simple::mtu))),
         _num_recv_frames(size_t(hints.cast<double>("num_recv_frames", DEFAULT_NUM_RECV_FRAMES))),
         _send_frame_size(size_t(hints.cast<double>("send_frame_size", udp_simple::mtu))),
-        _num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_SEND_FRAMES)))
+        _num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_SEND_FRAMES))),
+        _concurrency_hint(hints.cast<size_t>("concurrency_hint", CONCURRENCY_HINT)),
+        _io_service(_concurrency_hint)
     {
         //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl;
 
@@ -129,7 +137,7 @@ public:
 
         //spawn the service threads that will run the io service
         _work = new asio::io_service::work(_io_service); //new work to delete later
-        for (size_t i = 0; i < CONCURRENCY_HINT; i++) _thread_group.create_thread(
+        for (size_t i = 0; i < _concurrency_hint; i++) _thread_group.create_thread(
             boost::bind(&udp_zero_copy_asio_impl::service, this)
         );
     }
@@ -292,12 +300,6 @@ public:
     size_t get_send_frame_size(void) const {return _send_frame_size;}
 
 private:
-    //asio guts -> socket and service
-    asio::ip::udp::socket   *_socket;
-    asio::io_service        _io_service;
-    asio::io_service::work  *_work;
-    int                     _sock_fd;
-
     //memory management -> buffers and fifos
     boost::thread_group _thread_group;
     boost::shared_array<char> _send_buffer, _recv_buffer;
@@ -305,6 +307,13 @@ private:
     pending_buffs_type::sptr _pending_recv_buffs, _pending_send_buffs;
     const size_t _recv_frame_size, _num_recv_frames;
     const size_t _send_frame_size, _num_send_frames;
+
+    //asio guts -> socket and service
+    size_t                  _concurrency_hint;
+    asio::io_service        _io_service;
+    asio::ip::udp::socket   *_socket;
+    asio::io_service::work  *_work;
+    int                     _sock_fd;
 };
 
 /***********************************************************************
diff --git a/host/lib/transport/vrt_packet_handler.hpp b/host/lib/transport/vrt_packet_handler.hpp
index 939517411..278bcfeaa 100644
--- a/host/lib/transport/vrt_packet_handler.hpp
+++ b/host/lib/transport/vrt_packet_handler.hpp
@@ -318,7 +318,7 @@ template <typename T> UHD_INLINE T get_context_code(
     ){
         //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);
-        if_packet_info.packet_count = state.next_packet_seq++;
+        if_packet_info.packet_count = state.next_packet_seq;
 
         //get send buffers for each channel
         managed_send_buffs_t send_buffs(buffs.size()/chans_per_otw_buff);
@@ -345,6 +345,7 @@ template <typename T> UHD_INLINE T get_context_code(
             size_t num_bytes_total = (vrt_header_offset_words32+if_packet_info.num_packet_words32)*sizeof(boost::uint32_t);
             send_buffs[i]->commit(num_bytes_total);
         }
+        state.next_packet_seq++; //increment sequence after commits
         return num_samps;
     }
 
@@ -387,10 +388,19 @@ 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;
 
+            //TODO remove this code when sample counts of zero are supported by hardware
+            std::vector<const void *> buffs_(buffs);
+            size_t total_num_samps_(total_num_samps);
+            if (total_num_samps == 0){
+                static const boost::uint64_t zeros = 0; //max size of a host sample
+                buffs_ = std::vector<const void *>(buffs.size(), &zeros);
+                total_num_samps_ = 1;
+            }
+
             return _send1(
                 state,
-                buffs, 0,
-                std::min(total_num_samps, max_samples_per_packet),
+                buffs_, 0,
+                std::min(total_num_samps_, max_samples_per_packet),
                 if_packet_info,
                 io_type, otw_type,
                 vrt_packer,
diff --git a/host/lib/usrp/usrp2/fw_common.h b/host/lib/usrp/usrp2/fw_common.h
index e812e1221..2cd3ee595 100644
--- a/host/lib/usrp/usrp2/fw_common.h
+++ b/host/lib/usrp/usrp2/fw_common.h
@@ -33,8 +33,8 @@ extern "C" {
 #endif
 
 //fpga and firmware compatibility numbers
-#define USRP2_FPGA_COMPAT_NUM 2
-#define USRP2_FW_COMPAT_NUM 6
+#define USRP2_FPGA_COMPAT_NUM 3
+#define USRP2_FW_COMPAT_NUM 7
 
 //used to differentiate control packets over data port
 #define USRP2_INVALID_VRT_HEADER 0
diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp
index eba704059..8314cf9ae 100644
--- a/host/lib/usrp/usrp2/io_impl.cpp
+++ b/host/lib/usrp/usrp2/io_impl.cpp
@@ -18,11 +18,11 @@
 #include "../../transport/vrt_packet_handler.hpp"
 #include "usrp2_impl.hpp"
 #include "usrp2_regs.hpp"
+#include <uhd/utils/byteswap.hpp>
 #include <uhd/utils/thread_priority.hpp>
 #include <uhd/transport/convert_types.hpp>
 #include <uhd/transport/alignment_buffer.hpp>
 #include <boost/format.hpp>
-#include <boost/asio.hpp> //htonl and ntohl
 #include <boost/bind.hpp>
 #include <boost/thread.hpp>
 #include <iostream>
@@ -32,7 +32,73 @@ using namespace uhd::usrp;
 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;
+/***********************************************************************
+ * constants
+ **********************************************************************/
+static const int underflow_flags = 0
+    | async_metadata_t::EVENT_CODE_UNDERFLOW
+    | async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET
+;
+
+static const size_t vrt_send_header_offset_words32 = 1;
+
+/***********************************************************************
+ * flow control monitor for a single tx channel
+ *  - the pirate thread calls update
+ *  - the get send buffer calls check
+ **********************************************************************/
+class flow_control_monitor{
+public:
+    typedef boost::uint32_t seq_type;
+    typedef boost::shared_ptr<flow_control_monitor> sptr;
+
+    /*!
+     * Make a new flow control monitor.
+     * \param max_seqs_out num seqs before throttling
+     */
+    flow_control_monitor(seq_type max_seqs_out){
+        _last_seq_out = 0;
+        _last_seq_ack = 0;
+        _max_seqs_out = max_seqs_out;
+    }
+
+    /*!
+     * Check the flow control condition.
+     * \param seq the sequence to go out
+     * \param timeout the timeout in seconds
+     * \return false on timeout
+     */
+    UHD_INLINE bool check_fc_condition(seq_type seq, double timeout){
+        boost::this_thread::disable_interruption di; //disable because the wait can throw
+        boost::unique_lock<boost::mutex> lock(_fc_mutex);
+        _last_seq_out = seq;
+        return _fc_cond.timed_wait(
+            lock,
+            boost::posix_time::microseconds(long(timeout*1e6)),
+            boost::bind(&flow_control_monitor::ready, this)
+        );
+    }
+
+    /*!
+     * Update the flow control condition.
+     * \param seq the last sequence number to be ACK'd
+     */
+    UHD_INLINE void update_fc_condition(seq_type seq){
+        boost::unique_lock<boost::mutex> lock(_fc_mutex);
+        _last_seq_ack = seq;
+        lock.unlock();
+        _fc_cond.notify_one();
+    }
+
+private:
+    bool ready(void){
+        return seq_type(_last_seq_out -_last_seq_ack) < _max_seqs_out;
+    }
+
+    boost::mutex _fc_mutex;
+    boost::condition _fc_cond;
+    seq_type _last_seq_out, _last_seq_ack, _max_seqs_out;
+};
 
 /***********************************************************************
  * io impl details (internal to this file)
@@ -44,12 +110,14 @@ static const int underflow_flags = async_metadata_t::EVENT_CODE_UNDERFLOW | asyn
 struct usrp2_impl::io_impl{
     typedef alignment_buffer<managed_recv_buffer::sptr, time_spec_t> alignment_buffer_type;
 
-    io_impl(size_t num_frames, size_t width):
+    io_impl(size_t num_recv_frames, size_t send_frame_size, size_t width):
         packet_handler_recv_state(width),
-        recv_pirate_booty(alignment_buffer_type::make(num_frames, width)),
+        recv_pirate_booty(alignment_buffer_type::make(num_recv_frames, width)),
         async_msg_fifo(bounded_buffer<async_metadata_t>::make(100/*messages deep*/))
     {
-        /* NOP */
+        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))
+        );
     }
 
     ~io_impl(void){
@@ -63,6 +131,29 @@ struct usrp2_impl::io_impl{
         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,
+        double timeout
+    ){
+        UHD_ASSERT_THROW(trans.size() == buffs.size());
+
+        //calculate the flow control word
+        const boost::uint32_t fc_word32 = packet_handler_send_state.next_packet_seq;
+
+        //grab a managed buffer for each index
+        for (size_t i = 0; i < buffs.size(); i++){
+            if (not fc_mons[i]->check_fc_condition(fc_word32, timeout)) return false;
+            buffs[i] = trans[i]->get_send_buff(timeout);
+            if (not buffs[i].get()) return false;
+            buffs[i]->cast<boost::uint32_t *>()[0] = uhd::htonx(fc_word32);
+        }
+        return true;
+    }
+
+    //flow control monitors
+    std::vector<flow_control_monitor::sptr> fc_mons;
+
     //state management for the vrt packet handler code
     vrt_packet_handler::recv_state packet_handler_recv_state;
     vrt_packet_handler::send_state packet_handler_send_state;
@@ -112,8 +203,16 @@ void usrp2_impl::io_impl::recv_pirate_loop(
                 );
                 metadata.event_code = vrt_packet_handler::get_context_code<async_metadata_t::event_code_t>(vrt_hdr, if_packet_info);
 
+                //catch the flow control packets and react
+                if (metadata.event_code == 0){
+                    boost::uint32_t fc_word32 = (vrt_hdr + if_packet_info.num_header_words32)[1];
+                    this->fc_mons[index]->update_fc_condition(uhd::ntohx(fc_word32));
+                    continue;
+                }
+
                 //print the famous U, and push the metadata into the message queue
                 if (metadata.event_code & underflow_flags) std::cerr << "U" << std::flush;
+                //else std::cout << "metadata.event_code " << metadata.event_code << std::endl;
                 async_msg_fifo->push_with_pop_on_full(metadata);
                 continue;
             }
@@ -143,22 +242,13 @@ void usrp2_impl::io_impl::recv_pirate_loop(
  * Helper Functions
  **********************************************************************/
 void usrp2_impl::io_init(void){
-    //send a small data packet so the usrp2 knows the udp source port
-    BOOST_FOREACH(zero_copy_if::sptr data_transport, _data_transports){
-        managed_send_buffer::sptr send_buff = data_transport->get_send_buff();
-        static const boost::uint32_t data = htonl(USRP2_INVALID_VRT_HEADER);
-        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()){};
-    }
 
-    //the number of recv frames is the number for the first transport
     //the assumption is that all data transports should be identical
-    size_t num_frames = _data_transports.front()->get_num_recv_frames();
+    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_frames, _data_transports.size()));
+    _io_impl = UHD_PIMPL_MAKE(io_impl, (num_recv_frames, send_frame_size, _data_transports.size()));
 
     //create a new pirate thread for each zc if (yarr!!)
     for (size_t i = 0; i < _data_transports.size(); i++){
@@ -183,23 +273,10 @@ bool usrp2_impl::recv_async_msg(
 /***********************************************************************
  * Send Data
  **********************************************************************/
-static bool get_send_buffs(
-    const std::vector<udp_zero_copy::sptr> &trans,
-    vrt_packet_handler::managed_send_buffs_t &buffs,
-    double timeout
-){
-    UHD_ASSERT_THROW(trans.size() == buffs.size());
-    bool good = true;
-    for (size_t i = 0; i < buffs.size(); i++){
-        buffs[i] = trans[i]->get_send_buff(timeout);
-        good = good and (buffs[i].get() != NULL);
-    }
-    return good;
-}
-
 size_t usrp2_impl::get_max_send_samps_per_packet(void) const{
     static const size_t hdr_size = 0
         + vrt::max_if_hdr_words32*sizeof(boost::uint32_t)
+        + vrt_send_header_offset_words32*sizeof(boost::uint32_t)
         - sizeof(vrt::if_packet_info_t().cid) //no class id ever used
     ;
     const size_t bpp = _data_transports.front()->get_send_frame_size() - hdr_size;
@@ -218,8 +295,9 @@ size_t usrp2_impl::send(
         io_type, _tx_otw_type,                     //input and output types to convert
         _mboards.front()->get_master_clock_freq(), //master clock tick rate
         uhd::transport::vrt::if_hdr_pack_be,
-        boost::bind(&get_send_buffs, _data_transports, _1, timeout),
-        get_max_send_samps_per_packet()
+        boost::bind(&usrp2_impl::io_impl::get_send_buffs, _io_impl.get(), _data_transports, _1, timeout),
+        get_max_send_samps_per_packet(),
+        vrt_send_header_offset_words32
     );
 }
 
diff --git a/host/lib/usrp/usrp2/mboard_impl.cpp b/host/lib/usrp/usrp2/mboard_impl.cpp
index a0e6adfad..8f3ae5c1b 100644
--- a/host/lib/usrp/usrp2/mboard_impl.cpp
+++ b/host/lib/usrp/usrp2/mboard_impl.cpp
@@ -21,6 +21,7 @@
 #include <uhd/usrp/dsp_utils.hpp>
 #include <uhd/usrp/mboard_props.hpp>
 #include <uhd/utils/assert.hpp>
+#include <uhd/utils/byteswap.hpp>
 #include <uhd/utils/algorithm.hpp>
 #include <uhd/types/mac_addr.hpp>
 #include <uhd/types/dict.hpp>
@@ -38,11 +39,24 @@ using namespace uhd::usrp;
 usrp2_mboard_impl::usrp2_mboard_impl(
     size_t index,
     transport::udp_simple::sptr ctrl_transport,
-    size_t recv_frame_size
+    transport::zero_copy_if::sptr data_transport,
+    size_t recv_samps_per_packet,
+    const device_addr_t &flow_control_hints
 ):
     _index(index),
-    _recv_frame_size(recv_frame_size)
+    _recv_samps_per_packet(recv_samps_per_packet)
 {
+    //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();
+    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))
+    };
+    std::memcpy(send_buff->cast<void*>(), &data, sizeof(data));
+    send_buff->commit(sizeof(data));
+
     //make a new interface for usrp2 stuff
     _iface = usrp2_iface::make(ctrl_transport);
 
@@ -69,13 +83,8 @@ usrp2_mboard_impl::usrp2_mboard_impl(
         _allowed_decim_and_interp_rates.push_back(i);
     }
 
-    //Issue a stop streaming command (in case it was left running).
-    //Since this command is issued before the networking is setup,
-    //most if not all junk packets will never make it to the socket.
-    this->issue_ddc_stream_cmd(stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS);
-
     //init the rx control registers
-    _iface->poke32(U2_REG_RX_CTRL_NSAMPS_PER_PKT, _recv_frame_size);
+    _iface->poke32(U2_REG_RX_CTRL_NSAMPS_PER_PKT, _recv_samps_per_packet);
     _iface->poke32(U2_REG_RX_CTRL_NCHANNELS, 1);
     _iface->poke32(U2_REG_RX_CTRL_CLEAR_OVERRUN, 1); //reset
     _iface->poke32(U2_REG_RX_CTRL_VRT_HEADER, 0
@@ -94,6 +103,16 @@ usrp2_mboard_impl::usrp2_mboard_impl(
     _iface->poke32(U2_REG_TX_CTRL_REPORT_SID, 1);  //sid 1 (different from rx)
     _iface->poke32(U2_REG_TX_CTRL_POLICY, U2_FLAG_TX_CTRL_POLICY_NEXT_PACKET);
 
+    //setting the cycles per update
+    const double ups_per_sec = flow_control_hints.cast<double>("ups_per_sec", 100);
+    const size_t cycles_per_up = size_t(_clock_ctrl->get_master_clock_rate()/ups_per_sec);
+    _iface->poke32(U2_REG_TX_CTRL_CYCLES_PER_UP, U2_FLAG_TX_CTRL_UP_ENB | cycles_per_up);
+
+    //setting the packets per update
+    const double ups_per_fifo = flow_control_hints.cast<double>("ups_per_fifo", 8);
+    const size_t packets_per_up = size_t(usrp2_impl::sram_bytes/ups_per_fifo/data_transport->get_send_frame_size());
+    _iface->poke32(U2_REG_TX_CTRL_PACKETS_PER_UP, U2_FLAG_TX_CTRL_UP_ENB | packets_per_up);
+
     //init the ddc
     init_ddc_config();
 
@@ -115,7 +134,8 @@ usrp2_mboard_impl::usrp2_mboard_impl(
 }
 
 usrp2_mboard_impl::~usrp2_mboard_impl(void){
-    /* NOP */
+    _iface->poke32(U2_REG_TX_CTRL_CYCLES_PER_UP, 0);
+    _iface->poke32(U2_REG_TX_CTRL_PACKETS_PER_UP, 0);
 }
 
 /***********************************************************************
@@ -178,7 +198,7 @@ void usrp2_mboard_impl::set_time_spec(const time_spec_t &time_spec, bool now){
 
 void usrp2_mboard_impl::issue_ddc_stream_cmd(const stream_cmd_t &stream_cmd){
     _iface->poke32(U2_REG_RX_CTRL_STREAM_CMD, dsp_type1::calc_stream_cmd_word(
-        stream_cmd, _recv_frame_size
+        stream_cmd, _recv_samps_per_packet
     ));
     _iface->poke32(U2_REG_RX_CTRL_TIME_SECS,  boost::uint32_t(stream_cmd.time_spec.get_full_secs()));
     _iface->poke32(U2_REG_RX_CTRL_TIME_TICKS, stream_cmd.time_spec.get_tick_count(get_master_clock_freq()));
diff --git a/host/lib/usrp/usrp2/usrp2_iface.cpp b/host/lib/usrp/usrp2/usrp2_iface.cpp
index 2d450bfc6..55c42567e 100644
--- a/host/lib/usrp/usrp2/usrp2_iface.cpp
+++ b/host/lib/usrp/usrp2/usrp2_iface.cpp
@@ -30,18 +30,6 @@
 using namespace uhd;
 using namespace uhd::transport;
 
-/*!
- * FIXME: large timeout, ethernet pause frames...
- *
- * Use a large timeout to work-around the fact that
- * flow-control may throttle outgoing control packets
- * due to its use of ethernet pause frames.
- *
- * This will be fixed when host-based flow control is implemented,
- * along with larger incoming send buffers using the on-board SRAM.
- */
-static const size_t CONTROL_TIMEOUT_MS = 3000; //3 seconds
-
 class usrp2_iface_impl : public usrp2_iface{
 public:
 /***********************************************************************
@@ -187,7 +175,7 @@ public:
         boost::uint8_t usrp2_ctrl_data_in_mem[udp_simple::mtu]; //allocate max bytes for recv
         const usrp2_ctrl_data_t *ctrl_data_in = reinterpret_cast<const usrp2_ctrl_data_t *>(usrp2_ctrl_data_in_mem);
         while(true){
-            size_t len = _ctrl_transport->recv(boost::asio::buffer(usrp2_ctrl_data_in_mem), CONTROL_TIMEOUT_MS);
+            size_t len = _ctrl_transport->recv(boost::asio::buffer(usrp2_ctrl_data_in_mem));
             if(len >= sizeof(boost::uint32_t) and ntohl(ctrl_data_in->proto_ver) != USRP2_FW_COMPAT_NUM){
                 throw std::runtime_error(str(boost::format(
                     "Expected protocol compatibility number %d, but got %d:\n"
diff --git a/host/lib/usrp/usrp2/usrp2_impl.cpp b/host/lib/usrp/usrp2/usrp2_impl.cpp
index a680708ad..afc69f703 100644
--- a/host/lib/usrp/usrp2/usrp2_impl.cpp
+++ b/host/lib/usrp/usrp2/usrp2_impl.cpp
@@ -17,7 +17,7 @@
 
 #include "usrp2_impl.hpp"
 #include <uhd/transport/if_addrs.hpp>
-#include <uhd/transport/udp_simple.hpp>
+#include <uhd/transport/udp_zero_copy.hpp>
 #include <uhd/usrp/device_props.hpp>
 #include <uhd/utils/assert.hpp>
 #include <uhd/utils/static.hpp>
@@ -35,9 +35,6 @@ using namespace uhd::usrp;
 using namespace uhd::transport;
 namespace asio = boost::asio;
 
-//! wait this long for a control response when discovering devices
-static const size_t DISCOVERY_TIMEOUT_MS = 100;
-
 /***********************************************************************
  * Helper Functions
  **********************************************************************/
@@ -99,7 +96,7 @@ static uhd::device_addrs_t usrp2_find(const device_addr_t &hint){
     boost::uint8_t usrp2_ctrl_data_in_mem[udp_simple::mtu]; //allocate max bytes for recv
     const usrp2_ctrl_data_t *ctrl_data_in = reinterpret_cast<const usrp2_ctrl_data_t *>(usrp2_ctrl_data_in_mem);
     while(true){
-        size_t len = udp_transport->recv(asio::buffer(usrp2_ctrl_data_in_mem), DISCOVERY_TIMEOUT_MS);
+        size_t len = udp_transport->recv(asio::buffer(usrp2_ctrl_data_in_mem));
         //std::cout << len << "\n";
         if (len > offsetof(usrp2_ctrl_data_t, data)){
             //handle the received data
@@ -128,7 +125,7 @@ static device::sptr usrp2_make(const device_addr_t &device_addr){
 
     //create a ctrl and data transport for each address
     std::vector<udp_simple::sptr> ctrl_transports;
-    std::vector<udp_zero_copy::sptr> data_transports;
+    std::vector<zero_copy_if::sptr> data_transports;
 
     BOOST_FOREACH(const std::string &addr, std::split_string(device_addr["addr"])){
         ctrl_transports.push_back(udp_simple::make_connected(
@@ -141,7 +138,7 @@ static device::sptr usrp2_make(const device_addr_t &device_addr){
 
     //create the usrp2 implementation guts
     return device::sptr(
-        new usrp2_impl(ctrl_transports, data_transports)
+        new usrp2_impl(ctrl_transports, data_transports, device_addr)
     );
 }
 
@@ -154,7 +151,8 @@ UHD_STATIC_BLOCK(register_usrp2_device){
  **********************************************************************/
 usrp2_impl::usrp2_impl(
     std::vector<udp_simple::sptr> ctrl_transports,
-    std::vector<udp_zero_copy::sptr> data_transports
+    std::vector<zero_copy_if::sptr> data_transports,
+    const device_addr_t &flow_control_hints
 ):
     _data_transports(data_transports)
 {
@@ -173,7 +171,9 @@ usrp2_impl::usrp2_impl(
     //create a new mboard handler for each control transport
     for(size_t i = 0; i < ctrl_transports.size(); i++){
         _mboards.push_back(usrp2_mboard_impl::sptr(new usrp2_mboard_impl(
-            i, ctrl_transports[i], this->get_max_recv_samps_per_packet()
+            i, ctrl_transports[i], data_transports[i],
+            this->get_max_recv_samps_per_packet(),
+            flow_control_hints
         )));
         //use an empty name when there is only one mboard
         std::string name = (ctrl_transports.size() > 1)? boost::lexical_cast<std::string>(i) : "";
diff --git a/host/lib/usrp/usrp2/usrp2_impl.hpp b/host/lib/usrp/usrp2/usrp2_impl.hpp
index 558726a2b..2531bd6cb 100644
--- a/host/lib/usrp/usrp2/usrp2_impl.hpp
+++ b/host/lib/usrp/usrp2/usrp2_impl.hpp
@@ -33,7 +33,7 @@
 #include <boost/function.hpp>
 #include <uhd/transport/vrt_if_packet.hpp>
 #include <uhd/transport/udp_simple.hpp> //mtu
-#include <uhd/transport/udp_zero_copy.hpp>
+#include <uhd/transport/zero_copy.hpp>
 #include <uhd/usrp/dboard_manager.hpp>
 #include <uhd/usrp/subdev_spec.hpp>
 
@@ -84,7 +84,9 @@ public:
     usrp2_mboard_impl(
         size_t index,
         uhd::transport::udp_simple::sptr,
-        size_t recv_frame_size
+        uhd::transport::zero_copy_if::sptr,
+        size_t recv_samps_per_packet,
+        const uhd::device_addr_t &flow_control_hints
     );
     ~usrp2_mboard_impl(void);
 
@@ -95,7 +97,7 @@ public:
 private:
     size_t _index;
     int _rev_hi, _rev_lo;
-    const size_t _recv_frame_size;
+    const size_t _recv_samps_per_packet;
 
     //properties for this mboard
     void get(const wax::obj &, wax::obj &);
@@ -171,14 +173,18 @@ private:
  */
 class usrp2_impl : public uhd::device{
 public:
+    static const size_t sram_bytes = size_t(1 << 20);
+
     /*!
      * 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
      */
     usrp2_impl(
         std::vector<uhd::transport::udp_simple::sptr> ctrl_transports,
-        std::vector<uhd::transport::udp_zero_copy::sptr> data_transports
+        std::vector<uhd::transport::zero_copy_if::sptr> data_transports,
+        const uhd::device_addr_t &flow_control_hints
     );
 
     ~usrp2_impl(void);
@@ -208,7 +214,7 @@ private:
     uhd::dict<std::string, usrp2_mboard_impl::sptr> _mboard_dict;
 
     //io impl methods and members
-    std::vector<uhd::transport::udp_zero_copy::sptr> _data_transports;
+    std::vector<uhd::transport::zero_copy_if::sptr> _data_transports;
     uhd::otw_type_t _rx_otw_type, _tx_otw_type;
     UHD_PIMPL_DECL(io_impl) _io_impl;
     void io_init(void);
diff --git a/host/lib/usrp/usrp2/usrp2_regs.hpp b/host/lib/usrp/usrp2/usrp2_regs.hpp
index 064ad4e95..c3a4d22de 100644
--- a/host/lib/usrp/usrp2/usrp2_regs.hpp
+++ b/host/lib/usrp/usrp2/usrp2_regs.hpp
@@ -193,9 +193,14 @@
 #define U2_REG_TX_CTRL_CLEAR_STATE       _SR_ADDR(SR_TX_CTRL + 1)
 #define U2_REG_TX_CTRL_REPORT_SID        _SR_ADDR(SR_TX_CTRL + 2)
 #define U2_REG_TX_CTRL_POLICY            _SR_ADDR(SR_TX_CTRL + 3)
+#define U2_REG_TX_CTRL_CYCLES_PER_UP     _SR_ADDR(SR_TX_CTRL + 4)
+#define U2_REG_TX_CTRL_PACKETS_PER_UP    _SR_ADDR(SR_TX_CTRL + 5)
 
 #define U2_FLAG_TX_CTRL_POLICY_WAIT          (0x1 << 0)
 #define U2_FLAG_TX_CTRL_POLICY_NEXT_PACKET   (0x1 << 1)
 #define U2_FLAG_TX_CTRL_POLICY_NEXT_BURST    (0x1 << 2)
 
+//enable flag for registers: cycles and packets per update packet
+#define U2_FLAG_TX_CTRL_UP_ENB              (1ul << 31)
+
 #endif /* INCLUDED_USRP2_REGS_HPP */