lib: Remove boost::thread usage from USB transports

All the USB-based USRPs use boost::thread constructs, which are removed
with this commit. The following changes are applied:

- boost::condition is replaced with std::condition_variable
  - This requires shifting timeout values from boost::posix_time to
     std::chrono
  - boost::condition::timed_wait() is replaced by
    std::condition_variable::wait_for(), which returns a std::cv_status
    instead of bool.
- msg_task used a boost::barrier to check for the correct start of the
  underlying thread, which is removed to a simple check of the
  integrated boolean.

host/lib/transport/libusb1_zero_copy.cpp

@@ -12,9 +12,10 @@
 #include <uhd/transport/buffer_pool.hpp>
 #include <uhd/transport/usb_zero_copy.hpp>
 #include <uhd/utils/log.hpp>
+#include <condition_variable>
 #include <boost/circular_buffer.hpp>
 #include <boost/format.hpp>
-#include <boost/thread/condition.hpp>
+#include <chrono>
 #include <functional>
 #include <list>
 #include <memory>
@@ -50,7 +51,7 @@ struct lut_result_t
     libusb_transfer_status status;
     int actual_length;
     std::mutex mut;
-    boost::condition usb_transfer_complete;
+    std::condition_variable usb_transfer_complete;
 
 #ifdef UHD_TXRX_DEBUG_PRINTS
     // These are fore debugging
@@ -60,7 +61,7 @@ struct lut_result_t
 #endif
 };
 
-// Created to be used as an argument to boost::condition_variable::timed_wait() function
+// Created to be used as an argument to std::condition_variable::wait_for() function
 struct lut_result_completed
 {
     const lut_result_t& _result;
@@ -184,11 +185,9 @@ public:
             if (timeout < 0.0) {
                 result.usb_transfer_complete.wait(lock);
             } else {
-                const boost::system_time timeout_time =
+                result.usb_transfer_complete.wait_for(lock,
-                    boost::get_system_time()
+                    std::chrono::microseconds(static_cast<int64_t>(timeout * 1e6)),
-                    + boost::posix_time::microseconds(long(timeout * 1000000));
+                    lut_result_completed(result));
-                result.usb_transfer_complete.timed_wait(
-                    lock, timeout_time, lut_result_completed(result));
             }
         }
         return (result.completed > 0);
@@ -317,8 +316,8 @@ public:
 
         std::unique_lock<std::mutex> queue_lock(_queue_mutex);
         if (_enqueued.empty()) {
-            _buff_ready_cond.timed_wait(
+            _buff_ready_cond.wait_for(queue_lock,
-                queue_lock, boost::posix_time::microseconds(long(timeout * 1e6)));
+                std::chrono::microseconds(static_cast<int64_t>(timeout * 1e6)));
         }
         if (_enqueued.empty())
             return buff;
@@ -352,7 +351,7 @@ private:
     std::vector<std::shared_ptr<libusb_zero_copy_mb>> _mb_pool;
 
     std::mutex _queue_mutex;
-    boost::condition _buff_ready_cond;
+    std::condition_variable _buff_ready_cond;
     std::mutex _get_buff_mutex;
 
     //! why 2 queues? there is room in the future to have > N buffers but only N in flight

host/lib/usrp/b100/usb_zero_copy_wrapper.cpp

@@ -12,7 +12,7 @@
 #include <uhd/utils/log.hpp>
 #include <uhd/utils/tasks.hpp>
 #include <uhdlib/utils/atomic.hpp>
-#include <boost/thread/condition.hpp>
+#include <condition_variable>
 #include <functional>
 #include <memory>
 #include <mutex>
@@ -21,8 +21,7 @@
 using namespace uhd;
 using namespace uhd::transport;
 
-static const boost::posix_time::time_duration AUTOFLUSH_TIMEOUT(
+constexpr std::chrono::milliseconds AUTOFLUSH_TIMEOUT{1};
-    boost::posix_time::milliseconds(1));
 
 /***********************************************************************
  * USB zero copy wrapper - managed receive buffer
@@ -147,7 +146,7 @@ private:
 
     // private variables for auto flusher
     std::mutex _mutex;
-    boost::condition _cond;
+    std::condition_variable _cond;
     uhd::task::sptr _task;
     bool _ok_to_auto_flush;
 
@@ -158,7 +157,8 @@ private:
     void auto_flush(void)
     {
         std::unique_lock<std::mutex> lock(_mutex);
-        const bool timeout = not _cond.timed_wait(lock, AUTOFLUSH_TIMEOUT);
+        const bool timeout = _cond.wait_for(lock, AUTOFLUSH_TIMEOUT)
+                             == std::cv_status::timeout;
         if (timeout and _ok_to_auto_flush and _last_send_buff and _bytes_in_buffer != 0) {
             _last_send_buff->commit(_bytes_in_buffer);
             _last_send_buff.reset();

host/lib/usrp/usrp1/io_impl.cpp

@@ -640,7 +640,6 @@ double usrp1_impl::update_tx_dsp_freq(const size_t dspno, const double freq)
  **********************************************************************/
 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);
 }

host/lib/usrp/usrp1/soft_time_ctrl.cpp

@@ -8,7 +8,7 @@
 #include "soft_time_ctrl.hpp"
 #include <uhd/utils/tasks.hpp>
 #include <uhdlib/utils/system_time.hpp>
-#include <boost/thread/condition.hpp>
+#include <condition_variable>
 #include <chrono>
 #include <functional>
 #include <iostream>
@@ -71,10 +71,11 @@ public:
     UHD_INLINE void sleep_until_time(
         std::unique_lock<std::mutex>& lock, const time_spec_t& time)
     {
-        boost::condition cond;
+        std::condition_variable cond;
         // use a condition variable to unlock, sleep, lock
         const double seconds_to_sleep = (time - time_now()).get_real_secs();
-        cond.timed_wait(lock, pt::microseconds(long(seconds_to_sleep * 1e6)));
+        cond.wait_for(lock,
+            std::chrono::microseconds(static_cast<int64_t>(seconds_to_sleep * 1e6)));
     }
 
     /*******************************************************************

host/lib/usrp/usrp2/io_impl.cpp

@@ -19,9 +19,6 @@
 #include <uhdlib/asio.hpp>
 #include <uhdlib/usrp/common/async_packet_handler.hpp>
 #include <uhdlib/usrp/common/validate_subdev_spec.hpp>
-#include <boost/format.hpp>
-#include <boost/thread/mutex.hpp>
-#include <boost/thread/thread.hpp>
 #include <chrono>
 #include <functional>
 #include <iostream>
@@ -31,16 +28,6 @@
 using namespace uhd;
 using namespace uhd::usrp;
 using namespace uhd::transport;
-namespace asio = boost::asio;
-namespace pt   = boost::posix_time;
-
-/***********************************************************************
- * helpers
- **********************************************************************/
-static UHD_INLINE pt::time_duration to_time_dur(double timeout)
-{
-    return pt::microseconds(long(timeout * 1e6));
-}
 
 /***********************************************************************
  * constants
@@ -95,8 +82,9 @@ public:
         std::unique_lock<std::mutex> lock(_fc_mutex);
         if (this->ready())
             return true;
-        boost::this_thread::disable_interruption di; // disable because the wait can throw
+        return _fc_cond.wait_for(lock,
-        return _fc_cond.timed_wait(lock, to_time_dur(timeout), _ready_fcn);
+            std::chrono::microseconds(static_cast<int64_t>(timeout * 1e6)),
+            _ready_fcn);
     }
 
     /*!
@@ -118,7 +106,7 @@ private:
     }
 
     std::mutex _fc_mutex;
-    boost::condition _fc_cond;
+    std::condition_variable _fc_cond;
     seq_type _last_seq_out, _last_seq_ack;
     const seq_type _max_seqs_out;
     std::function<bool(void)> _ready_fcn;
@@ -388,7 +376,6 @@ void usrp2_impl::update_tx_subdev_spec(
  **********************************************************************/
 bool usrp2_impl::recv_async_msg(async_metadata_t& async_metadata, double timeout)
 {
-    boost::this_thread::disable_interruption di; // disable because the wait can throw
     return _io_impl->async_msg_fifo.pop_with_timed_wait(async_metadata, timeout);
 }
 
@@ -410,9 +397,9 @@ void usrp2_impl::program_stream_dest(
     // user has provided an alternative address and port for destination
     if (args.args.has_key("addr") and args.args.has_key("port")) {
         UHD_LOGGER_INFO("USRP2")
-            << boost::format("Programming streaming destination for custom address. "
+            << "Programming streaming destination for custom address. "
-                             "IPv4 Address: %s, UDP Port: %s")
+               "IPv4 Address: "
-                   % args.args["addr"] % args.args["port"];
+            << args.args["addr"] << ", UDP Port: " << args.args["port"];
 
         asio::io_context io_context;
         asio::ip::udp::resolver resolver(io_context);

host/lib/utils/tasks.cpp

@@ -10,8 +10,8 @@
 #include <uhd/utils/msg_task.hpp>
 #include <uhd/utils/tasks.hpp>
 #include <uhd/utils/thread.hpp>
-#include <boost/thread/barrier.hpp>
+#include <condition_variable>
-#include <boost/thread/thread.hpp>
+#include <boost/optional.hpp>
 #include <atomic>
 #include <exception>
 #include <functional>
@@ -85,18 +85,21 @@ msg_task::~msg_task(void)
 class msg_task_impl : public msg_task
 {
 public:
-    msg_task_impl(const task_fcn_type& task_fcn) : _spawn_barrier(2)
+    msg_task_impl(const task_fcn_type& task_fcn)
+        : _running(false), _thread([this, task_fcn]() { this->task_loop(task_fcn); })
     {
-        (void)_thread_group.create_thread(
+        // Wait for the thread to successfully start
-            std::bind(&msg_task_impl::task_loop, this, task_fcn));
+        std::unique_lock<std::mutex> _l(_mutex);
-        _spawn_barrier.wait();
+
+        _start_cv.wait(_l, [this]() { return _running.load(); });
     }
 
     ~msg_task_impl(void) override
     {
         _running = false;
-        _thread_group.interrupt_all();
+        if (_thread.joinable()) {
-        _thread_group.join_all();
+            _thread.join();
+        }
     }
 
     /*
@@ -123,7 +126,7 @@ private:
     void task_loop(const task_fcn_type& task_fcn)
     {
         _running = true;
-        _spawn_barrier.wait();
+        _start_cv.notify_one();
 
         try {
             while (_running) {
@@ -138,8 +141,6 @@ private:
                     _dump_queue.push_back(buff.get());
                 }
             }
-        } 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 (...) {
@@ -157,9 +158,9 @@ private:
     }
 
     std::mutex _mutex;
-    boost::thread_group _thread_group;
+    std::condition_variable _start_cv;
-    boost::barrier _spawn_barrier;
+    std::atomic<bool> _running;
-    bool _running;
+    std::thread _thread;
 
     /*
      * This queue holds stranded messages until a radio_ctrl_core grabs them via