uhd: tasks now use std::threads under the hood, and can't be interrupted

USRP1 and USRP2 used tasks that relied on Boost thread interruption
mechanisms. These were replaced with explicit atomics.

6 files changed, 23 insertions, 13 deletions

diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp
index 7ed1d8671..7cb38548f 100644
--- a/host/lib/usrp/usrp1/io_impl.cpp
+++ b/host/lib/usrp/usrp1/io_impl.cpp
@@ -32,6 +32,7 @@
 #include <boost/bind.hpp>
 #include <boost/format.hpp>
 #include <boost/make_shared.hpp>
+#include <atomic>
 
 #define bmFR_RX_FORMAT_SHIFT_SHIFT 0
 #define bmFR_RX_FORMAT_WIDTH_SHIFT 4
@@ -147,12 +148,14 @@ struct usrp1_impl::io_impl{
     io_impl(zero_copy_if::sptr data_transport):
         data_transport(data_transport),
         curr_buff(offset_send_buffer(data_transport->get_send_buff())),
-        omsb(boost::bind(&usrp1_impl::io_impl::commit_send_buff, this, _1, _2, _3))
+        omsb(boost::bind(&usrp1_impl::io_impl::commit_send_buff, this, _1, _2, _3)),
+        vandal_loop_exit(false)
     {
         /* NOP */
     }
 
     ~io_impl(void){
+        vandal_loop_exit = true;
         UHD_SAFE_CALL(flush_send_buff();)
     }
 
@@ -175,6 +178,7 @@ struct usrp1_impl::io_impl{
         return omsb.get_new(curr_buff, next_buff);
     }
 
+    std::atomic<bool> vandal_loop_exit;
     task::sptr vandal_task;
     boost::system_time last_send_time;
 };
@@ -247,7 +251,7 @@ void usrp1_impl::io_init(void){
 
     //create a new vandal thread to poll xerflow conditions
     _io_impl->vandal_task = task::make(boost::bind(
-        &usrp1_impl::vandal_conquest_loop, this
+        &usrp1_impl::vandal_conquest_loop, this, std::ref(_io_impl->vandal_loop_exit)
     ));
 }
 
@@ -271,7 +275,7 @@ void usrp1_impl::tx_stream_on_off(bool enb){
  * 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(void){
+void usrp1_impl::vandal_conquest_loop(std::atomic<bool> &exit_loop){
 
     //initialize the async metadata
     async_metadata_t async_metadata;
@@ -285,7 +289,7 @@ void usrp1_impl::vandal_conquest_loop(void){
     inline_metadata.error_code = rx_metadata_t::ERROR_CODE_OVERFLOW;
 
     //start the polling loop...
-    try{ while (not boost::this_thread::interruption_requested()){
+    try{ while (not exit_loop){
         uint8_t underflow = 0, overflow = 0;
 
         //shutoff transmit if it has been too long since send() was called
@@ -315,7 +319,6 @@ void usrp1_impl::vandal_conquest_loop(void){
 
         boost::this_thread::sleep(boost::posix_time::milliseconds(50));
     }}
-    catch(const boost::thread_interrupted &){} //normal exit condition
     catch(const std::exception &e){
         UHD_LOGGER_ERROR("USRP1") << "The vandal caught an unexpected exception " << e.what() ;
     }
diff --git a/host/lib/usrp/usrp1/soft_time_ctrl.cpp b/host/lib/usrp/usrp1/soft_time_ctrl.cpp
index bb8b3a704..9cef99a60 100644
--- a/host/lib/usrp/usrp1/soft_time_ctrl.cpp
+++ b/host/lib/usrp/usrp1/soft_time_ctrl.cpp
@@ -199,8 +199,9 @@ public:
 
     void recv_cmd_task(void){ //task is looped
         boost::shared_ptr<stream_cmd_t> cmd;
-        _cmd_queue.pop_with_wait(cmd);
-        recv_cmd_handle_cmd(*cmd);
+        if (_cmd_queue.pop_with_timed_wait(cmd, 0.25)) {
+            recv_cmd_handle_cmd(*cmd);
+        }
     }
 
     bounded_buffer<async_metadata_t> &get_async_queue(void){
diff --git a/host/lib/usrp/usrp1/usrp1_impl.hpp b/host/lib/usrp/usrp1/usrp1_impl.hpp
index 1aa255f8d..b45d138d1 100644
--- a/host/lib/usrp/usrp1/usrp1_impl.hpp
+++ b/host/lib/usrp/usrp1/usrp1_impl.hpp
@@ -33,6 +33,7 @@
 #include <uhd/transport/usb_zero_copy.hpp>
 #include <boost/weak_ptr.hpp>
 #include <complex>
+#include <atomic>
 
 #ifndef INCLUDED_USRP1_IMPL_HPP
 #define INCLUDED_USRP1_IMPL_HPP
@@ -144,7 +145,7 @@ private:
     bool has_rx_halfband(void);
     bool has_tx_halfband(void);
 
-    void vandal_conquest_loop(void);
+    void vandal_conquest_loop(std::atomic<bool> &);
 
     void set_reg(const std::pair<uint8_t, uint32_t> &reg);
 
diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp
index 992d70835..8f6a67453 100644
--- a/host/lib/usrp/usrp2/io_impl.cpp
+++ b/host/lib/usrp/usrp2/io_impl.cpp
@@ -165,7 +165,7 @@ struct usrp2_impl::io_impl{
     std::vector<flow_control_monitor::sptr> fc_mons;
 
     //methods and variables for the pirate crew
-    void recv_pirate_loop(zero_copy_if::sptr, size_t);
+    void recv_pirate_loop(zero_copy_if::sptr, size_t, const std::atomic<bool> &);
     std::list<task::sptr> pirate_tasks;
     bounded_buffer<async_metadata_t> async_msg_fifo;
     double tick_rate;
@@ -178,14 +178,14 @@ struct usrp2_impl::io_impl{
  * - put async message packets into queue
  **********************************************************************/
 void usrp2_impl::io_impl::recv_pirate_loop(
-    zero_copy_if::sptr err_xport, size_t index
+    zero_copy_if::sptr err_xport, size_t index, const std::atomic<bool> &exit_loop
 ){
     set_thread_priority_safe();
 
     //store a reference to the flow control monitor (offset by max dsps)
     flow_control_monitor &fc_mon = *(this->fc_mons[index]);
 
-    while (not boost::this_thread::interruption_requested()){
+    while (not exit_loop){
         managed_recv_buffer::sptr buff = err_xport->get_recv_buff();
         if (not buff.get()) continue; //ignore timeout/error buffers
 
@@ -252,7 +252,8 @@ void usrp2_impl::io_init(void){
         //spawn a new pirate to plunder the recv booty
         _io_impl->pirate_tasks.push_back(task::make(boost::bind(
             &usrp2_impl::io_impl::recv_pirate_loop, _io_impl.get(),
-            _mbc[mb].tx_dsp_xport, index++
+            _mbc[mb].tx_dsp_xport, index++,
+            boost::ref(_pirate_task_exit)
         )));
     }
 }
diff --git a/host/lib/usrp/usrp2/usrp2_impl.cpp b/host/lib/usrp/usrp2/usrp2_impl.cpp
index 78a9acb72..9ee13d289 100644
--- a/host/lib/usrp/usrp2/usrp2_impl.cpp
+++ b/host/lib/usrp/usrp2/usrp2_impl.cpp
@@ -316,7 +316,8 @@ static zero_copy_if::sptr make_xport(
  * Structors
  **********************************************************************/
 usrp2_impl::usrp2_impl(const device_addr_t &_device_addr) :
-    device_addr(_device_addr)
+    device_addr(_device_addr),
+    _pirate_task_exit(false)
 {
     UHD_LOGGER_INFO("USRP2") << "Opening a USRP2/N-Series device...";
 
@@ -786,6 +787,7 @@ usrp2_impl::usrp2_impl(const device_addr_t &_device_addr) :
 }
 
 usrp2_impl::~usrp2_impl(void){UHD_SAFE_CALL(
+    _pirate_task_exit = true;
     for(const std::string &mb:  _mbc.keys()){
         _mbc[mb].tx_dsp->set_updates(0, 0);
     }
diff --git a/host/lib/usrp/usrp2/usrp2_impl.hpp b/host/lib/usrp/usrp2/usrp2_impl.hpp
index 790daa749..087a4f8e9 100644
--- a/host/lib/usrp/usrp2/usrp2_impl.hpp
+++ b/host/lib/usrp/usrp2/usrp2_impl.hpp
@@ -46,6 +46,7 @@
 #include <uhd/usrp/dboard_manager.hpp>
 #include <uhd/usrp/subdev_spec.hpp>
 #include <boost/weak_ptr.hpp>
+#include <atomic>
 
 static const double USRP2_LINK_RATE_BPS = 1000e6/8;
 static const double mimo_clock_delay_usrp2_rev4 = 4.18e-9;
@@ -120,6 +121,7 @@ private:
     //io impl methods and members
     uhd::device_addr_t device_addr;
     UHD_PIMPL_DECL(io_impl) _io_impl;
+    std::atomic<bool> _pirate_task_exit;
     void io_init(void);
     void update_tick_rate(const double rate);
     void update_rx_samp_rate(const std::string &, const size_t, const double rate);