mirror of
https://github.com/EttusResearch/uhd.git
synced 2026-10-25 17:36:10 +00:00
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.
This commit is contained in:
parent
1edb3a4c0a
commit
3855516f6f
@ -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 =
|
||||
boost::get_system_time()
|
||||
+ boost::posix_time::microseconds(long(timeout * 1000000));
|
||||
result.usb_transfer_complete.timed_wait(
|
||||
lock, timeout_time, lut_result_completed(result));
|
||||
result.usb_transfer_complete.wait_for(lock,
|
||||
std::chrono::microseconds(static_cast<int64_t>(timeout * 1e6)),
|
||||
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(
|
||||
queue_lock, boost::posix_time::microseconds(long(timeout * 1e6)));
|
||||
_buff_ready_cond.wait_for(queue_lock,
|
||||
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
|
||||
|
||||
@ -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(
|
||||
boost::posix_time::milliseconds(1));
|
||||
constexpr std::chrono::milliseconds AUTOFLUSH_TIMEOUT{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();
|
||||
|
||||
@ -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);
|
||||
}
|
||||
|
||||
@ -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)));
|
||||
}
|
||||
|
||||
/*******************************************************************
|
||||
|
||||
@ -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.timed_wait(lock, to_time_dur(timeout), _ready_fcn);
|
||||
return _fc_cond.wait_for(lock,
|
||||
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. "
|
||||
"IPv4 Address: %s, UDP Port: %s")
|
||||
% args.args["addr"] % args.args["port"];
|
||||
<< "Programming streaming destination for custom address. "
|
||||
"IPv4 Address: "
|
||||
<< args.args["addr"] << ", UDP Port: " << args.args["port"];
|
||||
|
||||
asio::io_context io_context;
|
||||
asio::ip::udp::resolver resolver(io_context);
|
||||
|
||||
@ -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 <boost/thread/thread.hpp>
|
||||
#include <condition_variable>
|
||||
#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(
|
||||
std::bind(&msg_task_impl::task_loop, this, task_fcn));
|
||||
_spawn_barrier.wait();
|
||||
// Wait for the thread to successfully start
|
||||
std::unique_lock<std::mutex> _l(_mutex);
|
||||
|
||||
_start_cv.wait(_l, [this]() { return _running.load(); });
|
||||
}
|
||||
|
||||
~msg_task_impl(void) override
|
||||
{
|
||||
_running = false;
|
||||
_thread_group.interrupt_all();
|
||||
_thread_group.join_all();
|
||||
if (_thread.joinable()) {
|
||||
_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;
|
||||
boost::barrier _spawn_barrier;
|
||||
bool _running;
|
||||
std::condition_variable _start_cv;
|
||||
std::atomic<bool> _running;
|
||||
std::thread _thread;
|
||||
|
||||
/*
|
||||
* This queue holds stranded messages until a radio_ctrl_core grabs them via
|
||||
|
||||
Loading…
Reference in New Issue
Block a user