usrp1: tweaks + implemented other features to mimic async and inline messages

Moved the underflow/overflow polling into a thread and out of the fast-path.
Added an inline and async message queue into soft time control.
Error and status messages are actually generated now and enqueued.
Passes the async message test...

8 files changed, 171 insertions, 65 deletions

diff --git a/host/docs/usrp1.rst b/host/docs/usrp1.rst
index cf344d7e4..71b4f6779 100644
--- a/host/docs/usrp1.rst
+++ b/host/docs/usrp1.rst
@@ -74,6 +74,10 @@ List of emulated features
 * Receiving at a specific time
 * Receiving a specific number of samples
 * End of burst flags for transmit/receive
+* Notification on late stream command
+* Notification on late transmit packet
+* Notification on underflow or overflow
+* Notification on broken chain error
 
 **Note:**
 These emulated features rely on the host system's clock for timed operations,
@@ -82,10 +86,6 @@ and therefore may not have sufficient precision for the application.
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 List of missing features
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-* Notification on late stream command
-* Notification on late transmit packet
-* Notification on broken chain error
-* Notification on underflow or overflow
 * Start of burst flags for transmit/receive
 
 ------------------------------------------------------------------------
diff --git a/host/lib/transport/libusb1_control.cpp b/host/lib/transport/libusb1_control.cpp
index f903907d0..bce3d4b0b 100644
--- a/host/lib/transport/libusb1_control.cpp
+++ b/host/lib/transport/libusb1_control.cpp
@@ -17,6 +17,7 @@
 
 #include "libusb1_base.hpp"
 #include <uhd/transport/usb_control.hpp>
+#include <boost/thread/mutex.hpp>
 
 using namespace uhd::transport;
 
@@ -40,6 +41,7 @@ public:
                   unsigned char *buff,
                   boost::uint16_t length
     ){
+        boost::mutex::scoped_lock lock(_mutex);
         return libusb_control_transfer(_handle->get(),
                                        request_type,
                                        request,
@@ -52,6 +54,7 @@ public:
 
 private:
     libusb::device_handle::sptr _handle;
+    boost::mutex _mutex;
 };
 
 /***********************************************************************
diff --git a/host/lib/usrp/e100/io_impl.cpp b/host/lib/usrp/e100/io_impl.cpp
index 14b348f96..6e1bc3245 100644
--- a/host/lib/usrp/e100/io_impl.cpp
+++ b/host/lib/usrp/e100/io_impl.cpp
@@ -51,6 +51,11 @@ struct e100_impl::io_impl{
         false_alarm(0), async_msg_fifo(100/*messages deep*/)
     { /* NOP */ }
 
+    ~io_impl(void){
+        recv_pirate_crew.interrupt_all();
+        recv_pirate_crew.join_all();
+    }
+
     double tick_rate; //set by update tick rate method
     e100_ctrl::sptr iface; //so handle irq can peek and poke
     void handle_irq(void);
diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp
index 8e75f2025..461529ae2 100644
--- a/host/lib/usrp/usrp1/io_impl.cpp
+++ b/host/lib/usrp/usrp1/io_impl.cpp
@@ -28,6 +28,7 @@
 #include <uhd/transport/bounded_buffer.hpp>
 #include <boost/math/special_functions/sign.hpp>
 #include <boost/math/special_functions/round.hpp>
+#include <boost/thread/thread.hpp>
 #include <boost/bind.hpp>
 #include <boost/format.hpp>
 
@@ -123,8 +124,6 @@ static void usrp1_bs_vrt_unpacker(
 struct usrp1_impl::io_impl{
     io_impl(zero_copy_if::sptr data_transport):
         data_transport(data_transport),
-        underflow_poll_samp_count(0),
-        overflow_poll_samp_count(0),
         curr_buff(offset_send_buffer(data_transport->get_send_buff())),
         omsb(boost::bind(&usrp1_impl::io_impl::commit_send_buff, this, _1, _2, _3))
     {
@@ -132,6 +131,8 @@ struct usrp1_impl::io_impl{
     }
 
     ~io_impl(void){
+        vandal_tribe.interrupt_all();
+        vandal_tribe.join_all();
         UHD_SAFE_CALL(flush_send_buff();)
     }
 
@@ -141,12 +142,6 @@ struct usrp1_impl::io_impl{
     sph::recv_packet_handler recv_handler;
     sph::send_packet_handler send_handler;
 
-    //state management for overflow and underflow
-    size_t underflow_poll_samp_count;
-    size_t overflow_poll_samp_count;
-    size_t rx_samps_per_poll_interval;
-    size_t tx_samps_per_poll_interval;
-
     //wrapper around the actual send buffer interface
     //all of this to ensure only aligned lengths are committed
     //NOTE: you must commit before getting a new buffer
@@ -163,6 +158,9 @@ struct usrp1_impl::io_impl{
         //make a new managed buffer with the offset buffs
         return omsb.get_new(curr_buff, next_buff);
     }
+
+    boost::thread_group vandal_tribe;
+    boost::system_time last_send_time;
 };
 
 /*!
@@ -231,6 +229,14 @@ void usrp1_impl::io_init(void){
 
     _io_impl = UHD_PIMPL_MAKE(io_impl, (_data_transport));
 
+    //create a new vandal thread to poll xerflow conditions
+    boost::barrier spawn_barrier(2);
+    _io_impl->vandal_tribe.create_thread(boost::bind(
+        &usrp1_impl::vandal_conquest_loop,
+        this, boost::ref(spawn_barrier)
+    ));
+    spawn_barrier.wait();
+
     //init some handler stuff
     _io_impl->recv_handler.set_tick_rate(_master_clock_rate);
     _io_impl->recv_handler.set_vrt_unpacker(&usrp1_bs_vrt_unpacker);
@@ -243,23 +249,91 @@ void usrp1_impl::io_init(void){
         &usrp1_impl::io_impl::get_send_buff, _io_impl.get(), _1
     ));
 
-    this->enable_tx(true); //always enabled
+    //init as disabled, then call the real function (uses restore)
+    this->enable_rx(false);
+    this->enable_tx(false);
     rx_stream_on_off(false);
+    tx_stream_on_off(false);
     _io_impl->flush_send_buff();
 }
 
 void usrp1_impl::rx_stream_on_off(bool enb){
-    this->enable_rx(enb);
+    this->restore_rx(enb);
     //drain any junk in the receive transport after stop streaming command
     while(not enb and _data_transport->get_recv_buff().get() != NULL){
         /* NOP */
     }
 }
 
+void usrp1_impl::tx_stream_on_off(bool enb){
+    _io_impl->last_send_time = boost::get_system_time();
+    if (_tx_enabled and not enb) _io_impl->flush_send_buff();
+    this->restore_tx(enb);
+}
+
+/*!
+ * Casually poll the overflow and underflow registers.
+ * On an underflow, push an async message into the queue and print.
+ * On an overflow, interleave an inline message into recv and print.
+ * This procedure creates "soft" inline and async user messages.
+ */
+void usrp1_impl::vandal_conquest_loop(boost::barrier &spawn_barrier){
+    spawn_barrier.wait();
+
+    //initialize the async metadata
+    async_metadata_t async_metadata;
+    async_metadata.channel = 0;
+    async_metadata.has_time_spec = true;
+    async_metadata.event_code = async_metadata_t::EVENT_CODE_UNDERFLOW;
+
+    //initialize the inline metadata
+    rx_metadata_t inline_metadata;
+    inline_metadata.has_time_spec = true;
+    inline_metadata.error_code = rx_metadata_t::ERROR_CODE_OVERFLOW;
+
+    //start the polling loop...
+    try{ while (not boost::this_thread::interruption_requested()){
+        boost::uint8_t underflow = 0, overflow = 0;
+
+        //shutoff transmit if it has been too long since send() was called
+        if (_tx_enabled and (boost::get_system_time() - _io_impl->last_send_time) > boost::posix_time::milliseconds(100)){
+            this->tx_stream_on_off(false);
+        }
+
+        //always poll regardless of enabled so we can clear the conditions
+        _fx2_ctrl->usrp_control_read(
+            VRQ_GET_STATUS, 0, GS_TX_UNDERRUN, &underflow, sizeof(underflow)
+        );
+        _fx2_ctrl->usrp_control_read(
+            VRQ_GET_STATUS, 0, GS_RX_OVERRUN, &overflow, sizeof(overflow)
+        );
+
+        //handle message generation for xerflow conditions
+        if (_tx_enabled and underflow){
+            async_metadata.time_spec = _soft_time_ctrl->get_time();
+            _soft_time_ctrl->get_async_queue().push_with_pop_on_full(async_metadata);
+            UHD_MSG(fastpath) << "U";
+        }
+        if (_rx_enabled and overflow){
+            inline_metadata.time_spec = _soft_time_ctrl->get_time();
+            _soft_time_ctrl->get_inline_queue().push_with_pop_on_full(inline_metadata);
+            UHD_MSG(fastpath) << "O";
+        }
+
+        boost::this_thread::sleep(boost::posix_time::milliseconds(50));
+    }}
+    catch(const boost::thread_interrupted &){} //normal exit condition
+    catch(const std::exception &e){
+        UHD_MSG(error) << "The vandal caught an unexpected exception " << e.what() << std::endl;
+    }
+}
+
 /***********************************************************************
  * Properties callback methods below
  **********************************************************************/
 void usrp1_impl::update_rx_subdev_spec(const uhd::usrp::subdev_spec_t &spec){
+    boost::mutex::scoped_lock lock = _io_impl->recv_handler.get_scoped_lock();
+
     //sanity checking
     validate_subdev_spec(_tree, spec, "rx");
 
@@ -281,6 +355,8 @@ void usrp1_impl::update_rx_subdev_spec(const uhd::usrp::subdev_spec_t &spec){
 }
 
 void usrp1_impl::update_tx_subdev_spec(const uhd::usrp::subdev_spec_t &spec){
+    boost::mutex::scoped_lock lock = _io_impl->send_handler.get_scoped_lock();
+
     //sanity checking
     validate_subdev_spec(_tree, spec, "tx");
 
@@ -305,14 +381,13 @@ void usrp1_impl::update_tx_subdev_spec(const uhd::usrp::subdev_spec_t &spec){
 }
 
 double usrp1_impl::update_rx_samp_rate(const double samp_rate){
+    boost::mutex::scoped_lock lock = _io_impl->recv_handler.get_scoped_lock();
+
     size_t rate = boost::math::iround(_master_clock_rate / samp_rate);
 
     //clip the rate to something in range:
     rate = std::min<size_t>(std::max<size_t>(rate, 4), 256);
 
-    //TODO Poll every 100ms. Make it selectable?
-    _io_impl->rx_samps_per_poll_interval = size_t(0.1 * _master_clock_rate / rate);
-
     bool s = this->disable_rx();
     _iface->poke32(FR_DECIM_RATE, rate/2 - 1);
     this->restore_rx(s);
@@ -322,14 +397,13 @@ double usrp1_impl::update_rx_samp_rate(const double samp_rate){
 }
 
 double usrp1_impl::update_tx_samp_rate(const double samp_rate){
+    boost::mutex::scoped_lock lock = _io_impl->send_handler.get_scoped_lock();
+
     size_t rate = boost::math::iround(_master_clock_rate / samp_rate);
 
     //clip the rate to something in range:
     rate = std::min<size_t>(std::max<size_t>(rate, 4), 256);
 
-    //TODO Poll every 100ms. Make it selectable? 
-    _io_impl->tx_samps_per_poll_interval = size_t(0.1 * _master_clock_rate / rate);
-
     bool s = this->disable_tx();
     _iface->poke32(FR_INTERP_RATE, rate/2 - 1);
     this->restore_tx(s);
@@ -367,10 +441,11 @@ double usrp1_impl::update_tx_dsp_freq(const size_t dspno, const double freq){
 /***********************************************************************
  * Async Data
  **********************************************************************/
-bool usrp1_impl::recv_async_msg(uhd::async_metadata_t &, double timeout){
-    //dummy fill-in for the recv_async_msg
-    boost::this_thread::sleep(boost::posix_time::microseconds(long(timeout*1e6)));
-    return false;
+bool usrp1_impl::recv_async_msg(
+    async_metadata_t &async_metadata, double timeout
+){
+    boost::this_thread::disable_interruption di; //disable because the wait can throw
+    return _soft_time_ctrl->get_async_queue().pop_with_timed_wait(async_metadata, timeout);
 }
 
 /***********************************************************************
@@ -390,29 +465,23 @@ size_t usrp1_impl::send(
 ){
     if (_soft_time_ctrl->send_pre(metadata, timeout)) return 0;
 
+    this->tx_stream_on_off(true); //always enable (it will do the right thing)
     size_t num_samps_sent = _io_impl->send_handler.send(
         buffs, nsamps_per_buff,
         metadata, io_type,
         send_mode, timeout
     );
 
-    //handle eob flag (commit the buffer, disable the DACs)
+    //handle eob flag (commit the buffer, /*disable the DACs*/)
     //check num samps sent to avoid flush on incomplete/timeout
     if (metadata.end_of_burst and num_samps_sent == nsamps_per_buff){
-        _io_impl->flush_send_buff();
-    }
-
-    //handle the polling for underflow conditions
-    _io_impl->underflow_poll_samp_count += num_samps_sent;
-    if (_io_impl->underflow_poll_samp_count >= _io_impl->tx_samps_per_poll_interval){
-        _io_impl->underflow_poll_samp_count = 0; //reset count
-        boost::uint8_t underflow = 0;
-        int ret = _fx2_ctrl->usrp_control_read(
-            VRQ_GET_STATUS, 0, GS_TX_UNDERRUN,
-            &underflow, sizeof(underflow)
-        );
-        if (ret < 0)        UHD_MSG(error) << "USRP: underflow check failed" << std::endl;
-        else if (underflow) UHD_MSG(fastpath) << "U";
+        async_metadata_t metadata;
+        metadata.channel = 0;
+        metadata.has_time_spec = true;
+        metadata.time_spec = _soft_time_ctrl->get_time();
+        metadata.event_code = async_metadata_t::EVENT_CODE_BURST_ACK;
+        _soft_time_ctrl->get_async_queue().push_with_pop_on_full(metadata);
+        this->tx_stream_on_off(false);
     }
 
     return num_samps_sent;
@@ -433,6 +502,9 @@ size_t usrp1_impl::recv(
     rx_metadata_t &metadata, const io_type_t &io_type,
     recv_mode_t recv_mode, double timeout
 ){
+    //interleave a "soft" inline message into the receive stream:
+    if (_soft_time_ctrl->get_inline_queue().pop_with_haste(metadata)) return 0;
+
     size_t num_samps_recvd = _io_impl->recv_handler.recv(
         buffs, nsamps_per_buff,
         metadata, io_type,
@@ -441,18 +513,5 @@ size_t usrp1_impl::recv(
 
     _soft_time_ctrl->recv_post(metadata, num_samps_recvd);
 
-    //handle the polling for overflow conditions
-    _io_impl->overflow_poll_samp_count += num_samps_recvd;
-    if (_io_impl->overflow_poll_samp_count >= _io_impl->rx_samps_per_poll_interval){
-        _io_impl->overflow_poll_samp_count = 0; //reset count
-        boost::uint8_t overflow = 0;
-        int ret = _fx2_ctrl->usrp_control_read(
-            VRQ_GET_STATUS, 0, GS_RX_OVERRUN,
-            &overflow, sizeof(overflow)
-        );
-        if (ret < 0)       UHD_MSG(error) << "USRP: overflow check failed" << std::endl;
-        else if (overflow) UHD_MSG(fastpath) << "O";
-    }
-
     return num_samps_recvd;
 }
diff --git a/host/lib/usrp/usrp1/soft_time_ctrl.cpp b/host/lib/usrp/usrp1/soft_time_ctrl.cpp
index 1bab34e7b..cf602185d 100644
--- a/host/lib/usrp/usrp1/soft_time_ctrl.cpp
+++ b/host/lib/usrp/usrp1/soft_time_ctrl.cpp
@@ -16,8 +16,7 @@
 //
 
 #include "soft_time_ctrl.hpp"
-#include <uhd/transport/bounded_buffer.hpp>
-#include <boost/any.hpp>
+#include <boost/make_shared.hpp>
 #include <boost/thread/thread.hpp>
 #include <boost/thread/barrier.hpp>
 #include <boost/thread/condition_variable.hpp>
@@ -41,6 +40,8 @@ public:
         _nsamps_remaining(0),
         _stream_mode(stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS),
         _cmd_queue(2),
+        _async_msg_queue(100),
+        _inline_msg_queue(100),
         _stream_on_off(stream_on_off)
     {
         //synchronously spawn a new thread
@@ -102,11 +103,20 @@ public:
         //When to stop streaming:
         //The samples have been received and the stream mode is non-continuous.
         //Rewrite the sample count to clip to the requested number of samples.
-        if (_nsamps_remaining <= nsamps){
+        if (_nsamps_remaining <= nsamps) switch(_stream_mode){
+        case stream_cmd_t::STREAM_MODE_NUM_SAMPS_AND_MORE:{
+            rx_metadata_t metadata;
+            metadata.has_time_spec = true;
+            metadata.time_spec = this->time_now();
+            metadata.error_code = rx_metadata_t::ERROR_CODE_BROKEN_CHAIN;
+            _inline_msg_queue.push_with_pop_on_full(metadata);
+        } //continue to next case...
+        case stream_cmd_t::STREAM_MODE_NUM_SAMPS_AND_DONE:
             nsamps = _nsamps_remaining; //set nsamps, then stop
             md.end_of_burst = true;
             stream_on_off(false);
             return;
+        default: break;
         }
 
         //update the consumed samples
@@ -114,7 +124,7 @@ public:
     }
 
     void issue_stream_cmd(const stream_cmd_t &cmd){
-        _cmd_queue.push_with_wait(cmd);
+        _cmd_queue.push_with_wait(boost::make_shared<stream_cmd_t>(cmd));
     }
 
     void stream_on_off(bool enb){
@@ -134,7 +144,12 @@ public:
 
         //handle late packets
         if (time_at < time_now()){
-            //TODO post async message
+            async_metadata_t metadata;
+            metadata.channel = 0;
+            metadata.has_time_spec = true;
+            metadata.time_spec = this->time_now();
+            metadata.event_code = async_metadata_t::EVENT_CODE_TIME_ERROR;
+            _async_msg_queue.push_with_pop_on_full(metadata);
             return true;
         }
 
@@ -153,7 +168,12 @@ public:
         if (not cmd.stream_now){
             time_spec_t time_at(cmd.time_spec - TWIDDLE);
             if (time_at < time_now()){
-                //TODO inject late cmd inline error
+                rx_metadata_t metadata;
+                metadata.has_time_spec = true;
+                metadata.time_spec = this->time_now();
+                metadata.error_code = rx_metadata_t::ERROR_CODE_LATE_COMMAND;
+                _inline_msg_queue.push_with_pop_on_full(metadata);
+                _stream_mode = stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS;
             }
             else{
                 sleep_until_time(lock, time_at);
@@ -180,20 +200,30 @@ public:
     void recv_cmd_dispatcher(boost::barrier &spawn_barrier){
         spawn_barrier.wait();
         try{
-            boost::any cmd;
+            boost::shared_ptr<stream_cmd_t> cmd;
             while (true){
                 _cmd_queue.pop_with_wait(cmd);
-                recv_cmd_handle_cmd(boost::any_cast<stream_cmd_t>(cmd));
+                recv_cmd_handle_cmd(*cmd);
             }
         } catch(const boost::thread_interrupted &){}
     }
 
+    bounded_buffer<async_metadata_t> &get_async_queue(void){
+        return _async_msg_queue;
+    }
+
+    bounded_buffer<rx_metadata_t> &get_inline_queue(void){
+        return _inline_msg_queue;
+    }
+
 private:
     boost::mutex _update_mutex;
     size_t _nsamps_remaining;
     stream_cmd_t::stream_mode_t _stream_mode;
     time_spec_t _time_offset;
-    bounded_buffer<boost::any> _cmd_queue;
+    bounded_buffer<boost::shared_ptr<stream_cmd_t> > _cmd_queue;
+    bounded_buffer<async_metadata_t> _async_msg_queue;
+    bounded_buffer<rx_metadata_t> _inline_msg_queue;
     const cb_fcn_type _stream_on_off;
     boost::thread_group _thread_group;
 };
diff --git a/host/lib/usrp/usrp1/soft_time_ctrl.hpp b/host/lib/usrp/usrp1/soft_time_ctrl.hpp
index 7fdac7fc8..b2bf6c6f9 100644
--- a/host/lib/usrp/usrp1/soft_time_ctrl.hpp
+++ b/host/lib/usrp/usrp1/soft_time_ctrl.hpp
@@ -21,6 +21,7 @@
 #include <uhd/types/stream_cmd.hpp>
 #include <uhd/types/time_spec.hpp>
 #include <uhd/types/metadata.hpp>
+#include <uhd/transport/bounded_buffer.hpp>
 #include <boost/utility.hpp>
 #include <boost/shared_ptr.hpp>
 #include <boost/function.hpp>
@@ -45,8 +46,6 @@ public:
      * \return a new soft time control object
      */
     static sptr make(const cb_fcn_type &stream_on_off);
-        //TODO pass in the error queue for async msgs
-        //TODO pass in the queue for inline msgs
 
     //! Set the current time
     virtual void set_time(const time_spec_t &time) = 0;
@@ -62,6 +61,12 @@ public:
 
     //! Issue a stream command to receive
     virtual void issue_stream_cmd(const stream_cmd_t &cmd) = 0;
+
+    //! Get access to a buffer of async metadata
+    virtual transport::bounded_buffer<async_metadata_t> &get_async_queue(void) = 0;
+
+    //! Get access to a buffer of inline metadata
+    virtual transport::bounded_buffer<rx_metadata_t> &get_inline_queue(void) = 0;
 };
 
 }} //namespace
diff --git a/host/lib/usrp/usrp1/usrp1_impl.cpp b/host/lib/usrp/usrp1/usrp1_impl.cpp
index deb257a83..78137178b 100644
--- a/host/lib/usrp/usrp1/usrp1_impl.cpp
+++ b/host/lib/usrp/usrp1/usrp1_impl.cpp
@@ -294,8 +294,8 @@ usrp1_impl::usrp1_impl(const device_addr_t &device_addr){
             .coerce(boost::bind(&usrp1_impl::update_tx_samp_rate, this, _1));
         _tree->create<double>(tx_dsp_path / "freq/value")
             .coerce(boost::bind(&usrp1_impl::update_tx_dsp_freq, this, dspno, _1));
-        _tree->create<meta_range_t>(tx_dsp_path / "freq/range")
-            .set(meta_range_t(-_master_clock_rate/2, +_master_clock_rate/2));
+        _tree->create<meta_range_t>(tx_dsp_path / "freq/range") //magic scalar comes from codec control:
+            .set(meta_range_t(-_master_clock_rate*0.6875, +_master_clock_rate*0.6875));
     }
 
     ////////////////////////////////////////////////////////////////////
diff --git a/host/lib/usrp/usrp1/usrp1_impl.hpp b/host/lib/usrp/usrp1/usrp1_impl.hpp
index f156d0bc4..cb1497253 100644
--- a/host/lib/usrp/usrp1/usrp1_impl.hpp
+++ b/host/lib/usrp/usrp1/usrp1_impl.hpp
@@ -30,6 +30,7 @@
 #include <uhd/usrp/subdev_spec.hpp>
 #include <uhd/usrp/dboard_eeprom.hpp>
 #include <uhd/usrp/dboard_manager.hpp>
+#include <boost/thread/barrier.hpp>
 #include <uhd/transport/usb_zero_copy.hpp>
 
 #ifndef INCLUDED_USRP1_IMPL_HPP
@@ -116,6 +117,7 @@ private:
     UHD_PIMPL_DECL(io_impl) _io_impl;
     void io_init(void);
     void rx_stream_on_off(bool);
+    void tx_stream_on_off(bool);
     void handle_overrun(size_t);
 
     //otw types
@@ -128,6 +130,8 @@ private:
     bool has_rx_halfband(void);
     bool has_tx_halfband(void);
 
+    void vandal_conquest_loop(boost::barrier &);
+
     //handle the enables
     bool _rx_enabled, _tx_enabled;
     void enable_rx(bool enb){