uhd: added tasks to simplify thread spawning use cases

12 files changed, 190 insertions, 134 deletions

diff --git a/host/include/uhd/utils/CMakeLists.txt b/host/include/uhd/utils/CMakeLists.txt
index 88a0e612b..0bf98fb67 100644
--- a/host/include/uhd/utils/CMakeLists.txt
+++ b/host/include/uhd/utils/CMakeLists.txt
@@ -30,6 +30,7 @@ INSTALL(FILES
     safe_call.hpp
     safe_main.hpp
     static.hpp
+    tasks.hpp
     thread_priority.hpp
     DESTINATION ${INCLUDE_DIR}/uhd/utils
     COMPONENT headers
diff --git a/host/include/uhd/utils/tasks.hpp b/host/include/uhd/utils/tasks.hpp
new file mode 100644
index 000000000..38b2bddf0
--- /dev/null
+++ b/host/include/uhd/utils/tasks.hpp
@@ -0,0 +1,53 @@
+//
+// Copyright 2011 Ettus Research LLC
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+//
+
+#ifndef INCLUDED_UHD_UTILS_TASKS_HPP
+#define INCLUDED_UHD_UTILS_TASKS_HPP
+
+#include <uhd/config.hpp>
+#include <boost/shared_ptr.hpp>
+#include <boost/function.hpp>
+#include <boost/utility.hpp>
+
+namespace uhd{
+
+    class task : boost::noncopyable{
+    public:
+        typedef boost::shared_ptr<task> sptr;
+        typedef boost::function<void(void)> task_fcn_type;
+
+        /*!
+         * Create a new task object with function callback.
+         * The task function callback will be run in a loop.
+         * until the thread is interrupted by the deconstructor.
+         *
+         * A task should return in a reasonable amount of time
+         * or may block forever under the following conditions:
+         *  - The blocking call is interruptible.
+         *  - The task polls the interrupt condition.
+         *
+         * \param task_fcn the task callback function
+         * \return a new task object
+         */
+        static sptr make(const task_fcn_type &task_fcn);
+
+    };
+
+} //namespace uhd
+
+#endif /* INCLUDED_UHD_UTILS_TASKS_HPP */
+
diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index f781f890d..0fa856d34 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -21,11 +21,11 @@
 #include <uhd/transport/buffer_pool.hpp>
 #include <uhd/utils/thread_priority.hpp>
 #include <uhd/utils/msg.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <uhd/exception.hpp>
 #include <boost/function.hpp>
 #include <boost/foreach.hpp>
 #include <boost/thread/thread.hpp>
-#include <boost/thread/barrier.hpp>
 #include <list>
 
 using namespace uhd;
@@ -202,12 +202,10 @@ public:
         }
 
         //spawn the event handler threads
-        size_t concurrency = hints.cast<size_t>("concurrency_hint", 1);
-        boost::barrier spawn_barrier(concurrency+1);
-        for (size_t i = 0; i < concurrency; i++) _thread_group.create_thread(
-            boost::bind(&libusb_zero_copy_impl::run_event_loop, this, boost::ref(spawn_barrier))
-        );
-        spawn_barrier.wait();
+        const size_t concurrency = hints.cast<size_t>("concurrency_hint", 1);
+        for (size_t i = 0; i < concurrency; i++) _event_loop_tasks.push_back(task::make(
+            boost::bind(&libusb_zero_copy_impl::run_event_loop, this)
+        ));
     }
 
     ~libusb_zero_copy_impl(void){
@@ -221,9 +219,6 @@ public:
                 boost::this_thread::sleep(boost::posix_time::milliseconds(10));
             }
         }
-        //shutdown the threads
-        _thread_group.interrupt_all();
-        _thread_group.join_all();
     }
 
     managed_recv_buffer::sptr get_recv_buff(double timeout){
@@ -275,20 +270,17 @@ private:
     std::list<libusb_transfer *> _all_luts;
 
     //! event handler threads
-    boost::thread_group _thread_group;
+    std::list<task::sptr> _event_loop_tasks;
 
-    void run_event_loop(boost::barrier &spawn_barrier){
-        spawn_barrier.wait();
+    void run_event_loop(void){
         set_thread_priority_safe();
         libusb_context *context = libusb::session::get_global_session()->get_context();
-        try{
-            while (not boost::this_thread::interruption_requested()){
-                timeval tv;
-                tv.tv_sec = 0;
-                tv.tv_usec = 100000; //100ms
-                libusb_handle_events_timeout(context, &tv);
-            }
-        } catch(const boost::thread_interrupted &){}
+        while (not boost::this_thread::interruption_requested()){
+            timeval tv;
+            tv.tv_sec = 0;
+            tv.tv_usec = 100000; //100ms
+            libusb_handle_events_timeout(context, &tv);
+        }
     }
 
 };
diff --git a/host/lib/usrp/b100/b100_ctrl.cpp b/host/lib/usrp/b100/b100_ctrl.cpp
index 5b03fd591..e08b47ce4 100644
--- a/host/lib/usrp/b100/b100_ctrl.cpp
+++ b/host/lib/usrp/b100/b100_ctrl.cpp
@@ -22,10 +22,12 @@
 #include <uhd/transport/vrt_if_packet.hpp>
 #include <uhd/utils/thread_priority.hpp>
 #include <uhd/utils/msg.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <uhd/types/metadata.hpp>
 #include <uhd/types/serial.hpp>
 #include "ctrl_packet.hpp"
-#include <boost/thread.hpp>
+#include <boost/thread/thread.hpp>
+#include <boost/bind.hpp>
 #include <uhd/exception.hpp>
 
 using namespace uhd::transport;
@@ -40,19 +42,12 @@ public:
         _ctrl_transport(ctrl_transport),
         _seq(0)
     {
-        boost::barrier spawn_barrier(2);
-        viking_marauders.create_thread(boost::bind(&b100_ctrl_impl::viking_marauder_loop, this, boost::ref(spawn_barrier)));
-        spawn_barrier.wait();
+        viking_marauder = task::make(boost::bind(&b100_ctrl_impl::viking_marauder_loop, this));
     }
 
     int write(boost::uint32_t addr, const ctrl_data_t &data);
     ctrl_data_t read(boost::uint32_t addr, size_t len);
 
-    ~b100_ctrl_impl(void) {
-        viking_marauders.interrupt_all();
-        viking_marauders.join_all();
-    }
-
     bool get_ctrl_data(ctrl_data_t &pkt_data, double timeout);
 
     void poke32(wb_addr_type addr, boost::uint32_t data){
@@ -95,10 +90,10 @@ private:
     int send_pkt(boost::uint16_t *cmd);
 
     //änd hërë wë gö ä-Vïkïng för äsynchronous control packets
-    void viking_marauder_loop(boost::barrier &);
+    void viking_marauder_loop(void);
     bounded_buffer<ctrl_data_t> sync_ctrl_fifo;
     async_cb_type _async_cb;
-    boost::thread_group viking_marauders;
+    task::sptr viking_marauder;
 
     uhd::transport::zero_copy_if::sptr _ctrl_transport;
     boost::uint8_t _seq;
@@ -206,8 +201,7 @@ ctrl_data_t b100_ctrl_impl::read(boost::uint32_t addr, size_t len) {
  * never have more than 1 message in it, since it's expected that we'll
  * wait for a control operation to finish before starting another one.
  **********************************************************************/
-void b100_ctrl_impl::viking_marauder_loop(boost::barrier &spawn_barrier) {
-    spawn_barrier.wait();
+void b100_ctrl_impl::viking_marauder_loop(void){
     set_thread_priority_safe();
 
     while (not boost::this_thread::interruption_requested()){
diff --git a/host/lib/usrp/e100/io_impl.cpp b/host/lib/usrp/e100/io_impl.cpp
index 69ca214dc..a10b3ffb3 100644
--- a/host/lib/usrp/e100/io_impl.cpp
+++ b/host/lib/usrp/e100/io_impl.cpp
@@ -24,12 +24,13 @@
 #include "e100_regs.hpp"
 #include <uhd/utils/msg.hpp>
 #include <uhd/utils/log.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <uhd/utils/thread_priority.hpp>
 #include <uhd/transport/bounded_buffer.hpp>
 #include <boost/bind.hpp>
 #include <boost/format.hpp>
+#include <boost/bind.hpp>
 #include <boost/thread/thread.hpp>
-#include <boost/thread/barrier.hpp>
 #include <poll.h> //poll
 #include <fcntl.h> //open, close
 #include <sstream>
@@ -51,11 +52,6 @@ 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);
@@ -70,11 +66,8 @@ struct e100_impl::io_impl{
 
     //a pirate's life is the life for me!
     void recv_pirate_loop(
-        boost::barrier &spawn_barrier,
         spi_iface::sptr //keep a sptr to iface which shares gpio147
     ){
-        spawn_barrier.wait();
-
         //open the GPIO and set it up for an IRQ
         std::ofstream edge_file("/sys/class/gpio/gpio147/edge");
         edge_file << "rising" << std::endl << std::flush;
@@ -94,7 +87,7 @@ struct e100_impl::io_impl{
         ::close(fd);
     }
     bounded_buffer<async_metadata_t> async_msg_fifo;
-    boost::thread_group recv_pirate_crew;
+    task::sptr pirate_task;
 };
 
 void e100_impl::io_impl::handle_irq(void){
@@ -191,12 +184,9 @@ void e100_impl::io_init(void){
     _fpga_ctrl->poke32(E100_REG_SR_ERR_CTRL, 1 << 1); //start
 
     //spawn a pirate, yarrr!
-    boost::barrier spawn_barrier(2);
-    _io_impl->recv_pirate_crew.create_thread(boost::bind(
-        &e100_impl::io_impl::recv_pirate_loop, _io_impl.get(),
-        boost::ref(spawn_barrier), _aux_spi_iface
+    _io_impl->pirate_task = task::make(boost::bind(
+        &e100_impl::io_impl::recv_pirate_loop, _io_impl.get(), _aux_spi_iface
     ));
-    spawn_barrier.wait();
 
     //init some handler stuff
     _io_impl->recv_handler.set_vrt_unpacker(&vrt::if_hdr_unpack_le);
diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp
index b596bbd04..e81b00d1c 100644
--- a/host/lib/usrp/usrp1/io_impl.cpp
+++ b/host/lib/usrp/usrp1/io_impl.cpp
@@ -24,6 +24,7 @@
 #include "usrp_commands.h"
 #include "usrp1_impl.hpp"
 #include <uhd/utils/msg.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <uhd/utils/safe_call.hpp>
 #include <uhd/transport/bounded_buffer.hpp>
 #include <boost/math/special_functions/sign.hpp>
@@ -131,8 +132,6 @@ struct usrp1_impl::io_impl{
     }
 
     ~io_impl(void){
-        vandal_tribe.interrupt_all();
-        vandal_tribe.join_all();
         UHD_SAFE_CALL(flush_send_buff();)
     }
 
@@ -159,7 +158,7 @@ struct usrp1_impl::io_impl{
         return omsb.get_new(curr_buff, next_buff);
     }
 
-    boost::thread_group vandal_tribe;
+    task::sptr vandal_task;
     boost::system_time last_send_time;
 };
 
@@ -230,12 +229,9 @@ 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)
+    _io_impl->vandal_task = task::make(boost::bind(
+        &usrp1_impl::vandal_conquest_loop, this
     ));
-    spawn_barrier.wait();
 
     //init some handler stuff
     _io_impl->recv_handler.set_tick_rate(_master_clock_rate);
@@ -277,8 +273,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(boost::barrier &spawn_barrier){
-    spawn_barrier.wait();
+void usrp1_impl::vandal_conquest_loop(void){
 
     //initialize the async metadata
     async_metadata_t async_metadata;
diff --git a/host/lib/usrp/usrp1/soft_time_ctrl.cpp b/host/lib/usrp/usrp1/soft_time_ctrl.cpp
index ac0899e28..78481c3ff 100644
--- a/host/lib/usrp/usrp1/soft_time_ctrl.cpp
+++ b/host/lib/usrp/usrp1/soft_time_ctrl.cpp
@@ -16,9 +16,8 @@
 //
 
 #include "soft_time_ctrl.hpp"
+#include <uhd/utils/tasks.hpp>
 #include <boost/make_shared.hpp>
-#include <boost/thread/thread.hpp>
-#include <boost/thread/barrier.hpp>
 #include <boost/thread/condition_variable.hpp>
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include <iostream>
@@ -45,21 +44,12 @@ public:
         _stream_on_off(stream_on_off)
     {
         //synchronously spawn a new thread
-        boost::barrier spawn_barrier(2);
-        _thread_group.create_thread(boost::bind(
-            &soft_time_ctrl_impl::recv_cmd_dispatcher, this, boost::ref(spawn_barrier))
-        );
-        spawn_barrier.wait();
+        _recv_cmd_task = task::make(boost::bind(&soft_time_ctrl_impl::recv_cmd_task, this));
 
         //initialize the time to something
         this->set_time(time_spec_t(0.0));
     }
 
-    ~soft_time_ctrl_impl(void){
-        _thread_group.interrupt_all();
-        _thread_group.join_all();
-    }
-
     /*******************************************************************
      * Time control
      ******************************************************************/
@@ -204,15 +194,10 @@ public:
         _stream_mode = cmd.stream_mode;
     }
 
-    void recv_cmd_dispatcher(boost::barrier &spawn_barrier){
-        spawn_barrier.wait();
-        try{
-            boost::shared_ptr<stream_cmd_t> cmd;
-            while (true){
-                _cmd_queue.pop_with_wait(cmd);
-                recv_cmd_handle_cmd(*cmd);
-            }
-        } catch(const boost::thread_interrupted &){}
+    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);
     }
 
     bounded_buffer<async_metadata_t> &get_async_queue(void){
@@ -232,7 +217,7 @@ private:
     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;
+    task::sptr _recv_cmd_task;
 };
 
 /***********************************************************************
diff --git a/host/lib/usrp/usrp1/usrp1_impl.hpp b/host/lib/usrp/usrp1/usrp1_impl.hpp
index cb1497253..1fe0c1784 100644
--- a/host/lib/usrp/usrp1/usrp1_impl.hpp
+++ b/host/lib/usrp/usrp1/usrp1_impl.hpp
@@ -30,7 +30,6 @@
 #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
@@ -130,7 +129,7 @@ private:
     bool has_rx_halfband(void);
     bool has_tx_halfband(void);
 
-    void vandal_conquest_loop(boost::barrier &);
+    void vandal_conquest_loop(void);
 
     //handle the enables
     bool _rx_enabled, _tx_enabled;
diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp
index 4c55012ce..7028e1ff8 100644
--- a/host/lib/usrp/usrp2/io_impl.cpp
+++ b/host/lib/usrp/usrp2/io_impl.cpp
@@ -22,15 +22,15 @@
 #include "usrp2_regs.hpp"
 #include <uhd/utils/log.hpp>
 #include <uhd/utils/msg.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <uhd/exception.hpp>
 #include <uhd/utils/byteswap.hpp>
 #include <uhd/utils/thread_priority.hpp>
 #include <uhd/transport/bounded_buffer.hpp>
+#include <boost/thread/thread.hpp>
 #include <boost/format.hpp>
 #include <boost/bind.hpp>
 #include <boost/thread/mutex.hpp>
-#include <boost/thread/thread.hpp>
-#include <boost/thread/barrier.hpp>
 #include <iostream>
 
 using namespace uhd;
@@ -134,11 +134,6 @@ struct usrp2_impl::io_impl{
         /* NOP */
     }
 
-    ~io_impl(void){
-        recv_pirate_crew.interrupt_all();
-        recv_pirate_crew.join_all();
-    }
-
     managed_send_buffer::sptr get_send_buff(size_t chan, double timeout){
         flow_control_monitor &fc_mon = *fc_mons[chan];
 
@@ -163,8 +158,8 @@ struct usrp2_impl::io_impl{
     sph::send_packet_handler send_handler;
 
     //methods and variables for the pirate crew
-    void recv_pirate_loop(boost::barrier &, zero_copy_if::sptr, size_t);
-    boost::thread_group recv_pirate_crew;
+    void recv_pirate_loop(zero_copy_if::sptr, size_t);
+    std::list<task::sptr> pirate_tasks;
     bounded_buffer<async_metadata_t> async_msg_fifo;
     double tick_rate;
 };
@@ -176,11 +171,8 @@ struct usrp2_impl::io_impl{
  * - put async message packets into queue
  **********************************************************************/
 void usrp2_impl::io_impl::recv_pirate_loop(
-    boost::barrier &spawn_barrier,
-    zero_copy_if::sptr err_xport,
-    size_t index
+    zero_copy_if::sptr err_xport, size_t index
 ){
-    spawn_barrier.wait();
     set_thread_priority_safe();
 
     //store a reference to the flow control monitor (offset by max dsps)
@@ -231,7 +223,7 @@ void usrp2_impl::io_impl::recv_pirate_loop(
                 //TODO unknown received packet, may want to print error...
             }
         }catch(const std::exception &e){
-            UHD_MSG(error) << "Error (usrp2 recv pirate loop): " << e.what() << std::endl;
+            UHD_MSG(error) << "Error in recv pirate loop: " << e.what() << std::endl;
         }
     }
 }
@@ -264,17 +256,14 @@ void usrp2_impl::io_init(void){
     }
 
     //create a new pirate thread for each zc if (yarr!!)
-    boost::barrier spawn_barrier(_mbc.size()+1);
     size_t index = 0;
     BOOST_FOREACH(const std::string &mb, _mbc.keys()){
         //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(), boost::ref(spawn_barrier),
+        _io_impl->pirate_tasks.push_back(task::make(boost::bind(
+            &usrp2_impl::io_impl::recv_pirate_loop, _io_impl.get(),
             _mbc[mb].err_xports.at(0), index++
-        ));
+        )));
     }
-    spawn_barrier.wait();
 
     //init some handler stuff
     _io_impl->recv_handler.set_vrt_unpacker(&vrt::if_hdr_unpack_be);
diff --git a/host/lib/usrp/usrp2/usrp2_iface.cpp b/host/lib/usrp/usrp2/usrp2_iface.cpp
index 0db9e5d58..b1347119b 100644
--- a/host/lib/usrp/usrp2/usrp2_iface.cpp
+++ b/host/lib/usrp/usrp2/usrp2_iface.cpp
@@ -20,15 +20,15 @@
 #include "usrp2_iface.hpp"
 #include <uhd/exception.hpp>
 #include <uhd/utils/msg.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <uhd/types/dict.hpp>
 #include <boost/thread.hpp>
 #include <boost/foreach.hpp>
 #include <boost/asio.hpp> //used for htonl and ntohl
 #include <boost/assign/list_of.hpp>
 #include <boost/format.hpp>
+#include <boost/bind.hpp>
 #include <boost/tokenizer.hpp>
-#include <boost/thread/thread.hpp>
-#include <boost/thread/barrier.hpp>
 #include <boost/functional/hash.hpp>
 #include <algorithm>
 #include <iostream>
@@ -110,13 +110,12 @@ public:
 
     void lock_device(bool lock){
         if (lock){
-            boost::barrier spawn_barrier(2);
-            _lock_thread_group.create_thread(boost::bind(&usrp2_iface_impl::lock_loop, this, boost::ref(spawn_barrier)));
-            spawn_barrier.wait();
+            this->get_reg<boost::uint32_t, USRP2_REG_ACTION_FW_POKE32>(U2_FW_REG_LOCK_GPID, boost::uint32_t(get_gpid()));
+            _lock_task = task::make(boost::bind(&usrp2_iface_impl::lock_task, this));
         }
         else{
-            _lock_thread_group.interrupt_all();
-            _lock_thread_group.join_all();
+            _lock_task.reset(); //shutdown the task
+            this->get_reg<boost::uint32_t, USRP2_REG_ACTION_FW_POKE32>(U2_FW_REG_LOCK_TIME, 0); //unlock
         }
     }
 
@@ -132,29 +131,12 @@ public:
         return lock_gpid != boost::uint32_t(get_gpid());
     }
 
-    void lock_loop(boost::barrier &spawn_barrier){
-        spawn_barrier.wait();
-
-        try{
-            this->get_reg<boost::uint32_t, USRP2_REG_ACTION_FW_POKE32>(U2_FW_REG_LOCK_GPID, boost::uint32_t(get_gpid()));
-            while(true){
-                //re-lock in loop
-                boost::uint32_t curr_secs = this->peek32(U2_REG_TIME64_SECS_RB_IMM);
-                this->get_reg<boost::uint32_t, USRP2_REG_ACTION_FW_POKE32>(U2_FW_REG_LOCK_TIME, curr_secs);
-                //sleep for a bit
-                boost::this_thread::sleep(boost::posix_time::milliseconds(1500));
-            }
-        }
-        catch(const boost::thread_interrupted &){
-            this->get_reg<boost::uint32_t, USRP2_REG_ACTION_FW_POKE32>(U2_FW_REG_LOCK_TIME, 0); //unlock on exit
-        }
-        catch(const std::exception &e){
-            UHD_MSG(error)
-                << "An unexpected exception was caught in the locker loop." << std::endl
-                << "The device will automatically unlock from this process." << std::endl
-                << e.what() << std::endl
-            ;
-        }
+    void lock_task(void){
+        //re-lock in task
+        boost::uint32_t curr_secs = this->peek32(U2_REG_TIME64_SECS_RB_IMM);
+        this->get_reg<boost::uint32_t, USRP2_REG_ACTION_FW_POKE32>(U2_FW_REG_LOCK_TIME, curr_secs);
+        //sleep for a bit
+        boost::this_thread::sleep(boost::posix_time::milliseconds(1500));
     }
 
 /***********************************************************************
@@ -400,7 +382,7 @@ private:
     boost::uint32_t _protocol_compat;
 
     //lock thread stuff
-    boost::thread_group _lock_thread_group;
+    task::sptr _lock_task;
 };
 
 /***********************************************************************
diff --git a/host/lib/utils/CMakeLists.txt b/host/lib/utils/CMakeLists.txt
index c8268c7b0..fd3249099 100644
--- a/host/lib/utils/CMakeLists.txt
+++ b/host/lib/utils/CMakeLists.txt
@@ -136,5 +136,6 @@ LIBUHD_APPEND_SOURCES(
     ${CMAKE_CURRENT_SOURCE_DIR}/paths.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/props.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/static.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/tasks.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/thread_priority.cpp
 )
diff --git a/host/lib/utils/tasks.cpp b/host/lib/utils/tasks.cpp
new file mode 100644
index 000000000..ef56bb2de
--- /dev/null
+++ b/host/lib/utils/tasks.cpp
@@ -0,0 +1,75 @@
+//
+// Copyright 2011 Ettus Research LLC
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+//
+
+#include <uhd/utils/tasks.hpp>
+#include <uhd/utils/msg.hpp>
+#include <boost/thread/thread.hpp>
+#include <boost/thread/barrier.hpp>
+#include <exception>
+#include <iostream>
+
+using namespace uhd;
+
+class task_impl : public task{
+public:
+
+    task_impl(const task_fcn_type &task_fcn){
+        boost::barrier spawn_barrier(2);
+        _thread_group.create_thread(boost::bind(&task_impl::task_loop, this, task_fcn, boost::ref(spawn_barrier)));
+        spawn_barrier.wait();
+    }
+
+    ~task_impl(void){
+        _thread_group.interrupt_all();
+        _thread_group.join_all();
+    }
+
+private:
+
+    void task_loop(const task_fcn_type &task_fcn, boost::barrier &spawn_barrier){
+        spawn_barrier.wait();
+
+        try{
+            while (not boost::this_thread::interruption_requested()){
+                task_fcn();
+            }
+        }
+        catch(const boost::thread_interrupted &){
+            //this is an ok way to exit the task loop
+        }
+        catch(const std::exception &e){
+            do_error_msg(e.what());
+        }
+        catch(...){
+            do_error_msg("unknown exception");
+        }
+    }
+
+    void do_error_msg(const std::string &msg){
+        UHD_MSG(error)
+            << "An unexpected exception was caught in a task loop." << std::endl
+            << "The task loop will now exit, things may not work." << std::endl
+            << msg << std::endl
+        ;
+    }
+
+    boost::thread_group _thread_group;
+};
+
+task::sptr task::make(const task_fcn_type &task_fcn){
+    return task::sptr(new task_impl(task_fcn));
+}