aboutsummaryrefslogtreecommitdiffstats
path: root/host/lib/usrp/device3/device3_io_impl.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'host/lib/usrp/device3/device3_io_impl.cpp')
-rw-r--r--host/lib/usrp/device3/device3_io_impl.cpp146
1 files changed, 76 insertions, 70 deletions
diff --git a/host/lib/usrp/device3/device3_io_impl.cpp b/host/lib/usrp/device3/device3_io_impl.cpp
index 198ee4022..374232972 100644
--- a/host/lib/usrp/device3/device3_io_impl.cpp
+++ b/host/lib/usrp/device3/device3_io_impl.cpp
@@ -23,7 +23,7 @@
#include <uhd/rfnoc/sink_block_ctrl_base.hpp>
#include <uhd/utils/byteswap.hpp>
#include <uhd/utils/log.hpp>
-#include <uhd/utils/msg.hpp>
+
#include "../common/async_packet_handler.hpp"
#include "../../transport/super_recv_packet_handler.hpp"
#include "../../transport/super_send_packet_handler.hpp"
@@ -34,7 +34,8 @@
#include <uhd/transport/zero_copy_flow_ctrl.hpp>
#include <boost/atomic.hpp>
-#define UHD_STREAMER_LOG() UHD_LOGV(never)
+#define UHD_TX_STREAMER_LOG() UHD_LOGGER_TRACE("STREAMER")
+#define UHD_RX_STREAMER_LOG() UHD_LOGGER_TRACE("STREAMER")
using namespace uhd;
using namespace uhd::usrp;
@@ -140,7 +141,7 @@ void generate_channel_list(
}
// Add all remaining args to all channel args
- BOOST_FOREACH(device_addr_t &chan_arg, chan_args_) {
+ for(device_addr_t &chan_arg: chan_args_) {
chan_arg = chan_arg.to_string() + "," + args.args.to_string();
}
@@ -251,7 +252,7 @@ static void handle_rx_flowctrl(
// Super-verbose mode:
//static size_t fc_pkt_count = 0;
- //UHD_MSG(status) << "sending flow ctrl packet " << fc_pkt_count++ << ", acking " << str(boost::format("%04d\tseq_sw==0x%08x") % last_seq % seq32) << std::endl;
+ //UHD_LOGGER_INFO("STREAMER") << "sending flow ctrl packet " << fc_pkt_count++ << ", acking " << str(boost::format("%04d\tseq_sw==0x%08x") % last_seq % seq32) ;
//load packet info
vrt::if_packet_info_t packet_info;
@@ -336,7 +337,7 @@ static size_t get_tx_flow_control_window(
static bool tx_flow_ctrl(
boost::shared_ptr<tx_fc_cache_t> fc_cache,
- zero_copy_if::sptr async_xport,
+ zero_copy_if::sptr async_xport,
uint32_t (*endian_conv)(uint32_t),
void (*unpack)(const uint32_t *packet_buff, vrt::if_packet_info_t &),
managed_buffer::sptr
@@ -351,35 +352,38 @@ static bool tx_flow_ctrl(
return true;
}
- // Look for a flow control message to update the space available in the buffer.
+ // Look for a flow control message to update the space available in the buffer.
// A minimal timeout is used because larger timeouts can cause the thread to be
// scheduled out for too long at high data rates and result in underruns.
- managed_recv_buffer::sptr buff = async_xport->get_recv_buff(0.000001);
- if (buff)
- {
- vrt::if_packet_info_t if_packet_info;
- if_packet_info.num_packet_words32 = buff->size()/sizeof(uint32_t);
- const uint32_t *packet_buff = buff->cast<const uint32_t *>();
- try {
- unpack(packet_buff, if_packet_info);
- }
- catch(const std::exception &ex)
- {
- UHD_MSG(error) << "Error unpacking async flow control packet: " << ex.what() << std::endl;
- continue;
- }
-
- if (if_packet_info.packet_type != vrt::if_packet_info_t::PACKET_TYPE_FC)
- {
- UHD_MSG(error) << "Unexpected packet type received by flow control handler: " << if_packet_info.packet_type << std::endl;
- continue;
- }
-
- // update the amount of space
- size_t seq_ack = endian_conv(packet_buff[if_packet_info.num_header_words32+1]);
- fc_cache->space += (seq_ack - fc_cache->last_seq_ack) & HW_SEQ_NUM_MASK;
- fc_cache->last_seq_ack = seq_ack;
- }
+ managed_recv_buffer::sptr buff = async_xport->get_recv_buff(0.000001);
+ if (buff)
+ {
+ vrt::if_packet_info_t if_packet_info;
+ if_packet_info.num_packet_words32 = buff->size()/sizeof(uint32_t);
+ const uint32_t *packet_buff = buff->cast<const uint32_t *>();
+ try {
+ unpack(packet_buff, if_packet_info);
+ }
+ catch(const std::exception &ex)
+ {
+ UHD_LOG_ERROR("TX FLOW CTRL", "Error unpacking async flow control packet: " << ex.what());
+ continue;
+ }
+
+ if (if_packet_info.packet_type != vrt::if_packet_info_t::PACKET_TYPE_FC)
+ {
+ UHD_LOG_ERROR(
+ "TX FLOW CTRL",
+ "Unexpected packet type received by flow control handler: " << if_packet_info.packet_type
+ );
+ continue;
+ }
+
+ // update the amount of space
+ size_t seq_ack = endian_conv(packet_buff[if_packet_info.num_header_words32+1]);
+ fc_cache->space += (seq_ack - fc_cache->last_seq_ack) & HW_SEQ_NUM_MASK;
+ fc_cache->last_seq_ack = seq_ack;
+ }
}
return false;
}
@@ -434,7 +438,7 @@ static void handle_tx_async_msgs(
}
catch(const std::exception &ex)
{
- UHD_MSG(error) << "Error parsing async message packet: " << ex.what() << std::endl;
+ UHD_LOGGER_ERROR("STREAMER") << "Error parsing async message packet: " << ex.what() ;
return;
}
@@ -454,10 +458,13 @@ static void handle_tx_async_msgs(
async_info->stream_channel
);
- // Filter out any flow control messages and cache the rest
+ // Filter out any flow control messages and cache the rest
if (metadata.event_code == DEVICE3_ASYNC_EVENT_CODE_FLOW_CTRL)
{
- UHD_MSG(error) << "Unexpected flow control message found in async message handling" << std::endl;
+ UHD_LOG_ERROR(
+ "TX ASYNC",
+ "Unexpected flow control message found in async message handling"
+ );
} else {
async_info->async_queue->push_with_pop_on_full(metadata);
metadata.channel = async_info->device_channel;
@@ -478,8 +485,8 @@ bool device3_impl::recv_async_msg(
**********************************************************************/
void device3_impl::update_rx_streamers(double /* rate */)
{
- BOOST_FOREACH(const std::string &block_id, _rx_streamers.keys()) {
- UHD_STREAMER_LOG() << "[Device3] updating RX streamer to " << block_id << std::endl;
+ for(const std::string &block_id: _rx_streamers.keys()) {
+ UHD_RX_STREAMER_LOG() << "updating RX streamer to " << block_id;
boost::shared_ptr<sph::recv_packet_streamer> my_streamer =
boost::dynamic_pointer_cast<sph::recv_packet_streamer>(_rx_streamers[block_id].lock());
if (my_streamer) {
@@ -496,7 +503,7 @@ void device3_impl::update_rx_streamers(double /* rate */)
if (scaling == rfnoc::scalar_node_ctrl::SCALE_UNDEFINED) {
scaling = 1/32767.;
}
- UHD_STREAMER_LOG() << " New tick_rate == " << tick_rate << " New samp_rate == " << samp_rate << " New scaling == " << scaling << std::endl;
+ UHD_RX_STREAMER_LOG() << " New tick_rate == " << tick_rate << " New samp_rate == " << samp_rate << " New scaling == " << scaling ;
my_streamer->set_tick_rate(tick_rate);
my_streamer->set_samp_rate(samp_rate);
@@ -525,7 +532,7 @@ rx_streamer::sptr device3_impl::get_rx_stream(const stream_args_t &args_)
for (size_t stream_i = 0; stream_i < chan_list.size(); stream_i++) {
// Get block ID and mb index
uhd::rfnoc::block_id_t block_id = chan_list[stream_i];
- UHD_STREAMER_LOG() << "[RX Streamer] chan " << stream_i << " connecting to " << block_id << std::endl;
+ UHD_RX_STREAMER_LOG() << "chan " << stream_i << " connecting to " << block_id ;
// Update args so args.args is always valid for this particular channel:
args.args = chan_args[stream_i];
size_t mb_index = block_id.get_device_no();
@@ -553,20 +560,20 @@ rx_streamer::sptr device3_impl::get_rx_stream(const stream_args_t &args_)
//allocate sid and create transport
uhd::sid_t stream_address = blk_ctrl->get_address(block_port);
- UHD_STREAMER_LOG() << "[RX Streamer] creating rx stream " << rx_hints.to_string() << std::endl;
+ UHD_RX_STREAMER_LOG() << "creating rx stream " << rx_hints.to_string() ;
both_xports_t xport = make_transport(stream_address, RX_DATA, rx_hints);
- UHD_STREAMER_LOG() << std::hex << "[RX Streamer] data_sid = " << xport.send_sid << std::dec << " actual recv_buff_size = " << xport.recv_buff_size << std::endl;
+ UHD_RX_STREAMER_LOG() << std::hex << "data_sid = " << xport.send_sid << std::dec << " actual recv_buff_size = " << xport.recv_buff_size ;
// Configure the block
blk_ctrl->set_destination(xport.send_sid.get_src(), block_port);
blk_ctrl->sr_write(uhd::rfnoc::SR_RESP_OUT_DST_SID, xport.send_sid.get_src(), block_port);
- UHD_STREAMER_LOG() << "[RX Streamer] resp_out_dst_sid == " << xport.send_sid.get_src() << std::endl;
+ UHD_RX_STREAMER_LOG() << "resp_out_dst_sid == " << xport.send_sid.get_src() ;
// Find all upstream radio nodes and set their response in SID to the host
std::vector<boost::shared_ptr<uhd::rfnoc::radio_ctrl> > upstream_radio_nodes = blk_ctrl->find_upstream_node<uhd::rfnoc::radio_ctrl>();
- UHD_STREAMER_LOG() << "[RX Streamer] Number of upstream radio nodes: " << upstream_radio_nodes.size() << std::endl;
- BOOST_FOREACH(const boost::shared_ptr<uhd::rfnoc::radio_ctrl> &node, upstream_radio_nodes) {
+ UHD_RX_STREAMER_LOG() << "Number of upstream radio nodes: " << upstream_radio_nodes.size();
+ for(const boost::shared_ptr<uhd::rfnoc::radio_ctrl> &node: upstream_radio_nodes) {
node->sr_write(uhd::rfnoc::SR_RESP_OUT_DST_SID, xport.send_sid.get_src(), block_port);
}
@@ -575,7 +582,7 @@ rx_streamer::sptr device3_impl::get_rx_stream(const stream_args_t &args_)
const size_t bpp = xport.recv->get_recv_frame_size() - stream_options.rx_max_len_hdr; // bytes per packet
const size_t bpi = convert::get_bytes_per_item(args.otw_format); // bytes per item
const size_t spp = std::min(args.args.cast<size_t>("spp", bpp/bpi), bpp/bpi); // samples per packet
- UHD_STREAMER_LOG() << "[RX Streamer] spp == " << spp << std::endl;
+ UHD_RX_STREAMER_LOG() << "spp == " << spp ;
//make the new streamer given the samples per packet
if (not my_streamer)
@@ -584,7 +591,7 @@ rx_streamer::sptr device3_impl::get_rx_stream(const stream_args_t &args_)
//init some streamer stuff
std::string conv_endianness;
- if (get_transport_endianness(mb_index) == ENDIANNESS_BIG) {
+ if (xport.endianness == ENDIANNESS_BIG) {
my_streamer->set_vrt_unpacker(&vrt::chdr::if_hdr_unpack_be);
conv_endianness = "be";
} else {
@@ -604,7 +611,7 @@ rx_streamer::sptr device3_impl::get_rx_stream(const stream_args_t &args_)
const size_t pkt_size = spp * bpi + stream_options.rx_max_len_hdr;
const size_t fc_window = get_rx_flow_control_window(pkt_size, xport.recv_buff_size, rx_hints);
const size_t fc_handle_window = std::max<size_t>(1, fc_window / stream_options.rx_fc_request_freq);
- UHD_STREAMER_LOG()<< "[RX Streamer] Flow Control Window (minus one) = " << fc_window-1 << ", Flow Control Handler Window = " << fc_handle_window << std::endl;
+ UHD_RX_STREAMER_LOG()<< "Flow Control Window (minus one) = " << fc_window-1 << ", Flow Control Handler Window = " << fc_handle_window ;
blk_ctrl->configure_flow_control_out(
fc_window-1, // Leave one space for overrun packets TODO make this obsolete
block_port
@@ -637,7 +644,7 @@ rx_streamer::sptr device3_impl::get_rx_stream(const stream_args_t &args_)
&handle_rx_flowctrl,
xport.send_sid,
xport.send,
- get_transport_endianness(mb_index),
+ xport.endianness,
fc_cache,
_1
),
@@ -680,8 +687,8 @@ rx_streamer::sptr device3_impl::get_rx_stream(const stream_args_t &args_)
**********************************************************************/
void device3_impl::update_tx_streamers(double /* rate */)
{
- BOOST_FOREACH(const std::string &block_id, _tx_streamers.keys()) {
- UHD_STREAMER_LOG() << "[Device3] updating TX streamer: " << block_id << std::endl;
+ for(const std::string &block_id: _tx_streamers.keys()) {
+ UHD_TX_STREAMER_LOG() << "updating TX streamer: " << block_id;
boost::shared_ptr<sph::send_packet_streamer> my_streamer =
boost::dynamic_pointer_cast<sph::send_packet_streamer>(_tx_streamers[block_id].lock());
if (my_streamer) {
@@ -697,7 +704,7 @@ void device3_impl::update_tx_streamers(double /* rate */)
if (scaling == rfnoc::scalar_node_ctrl::SCALE_UNDEFINED) {
scaling = 32767.;
}
- UHD_STREAMER_LOG() << " New tick_rate == " << tick_rate << " New samp_rate == " << samp_rate << " New scaling == " << scaling << std::endl;
+ UHD_TX_STREAMER_LOG() << "New tick_rate == " << tick_rate << " New samp_rate == " << samp_rate << " New scaling == " << scaling ;
my_streamer->set_tick_rate(tick_rate);
my_streamer->set_samp_rate(samp_rate);
my_streamer->set_scale_factor(scaling);
@@ -746,7 +753,6 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
args.args = chan_args[stream_i];
size_t mb_index = block_id.get_device_no();
size_t suggested_block_port = args.args.cast<size_t>("block_port", rfnoc::ANY_PORT);
- uhd::endianness_t endianness = get_transport_endianness(mb_index);
// Access to this channel's block control
uhd::rfnoc::sink_block_ctrl_base::sptr blk_ctrl =
@@ -771,17 +777,17 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
//allocate sid and create transport
uhd::sid_t stream_address = blk_ctrl->get_address(block_port);
- UHD_STREAMER_LOG() << "[TX Streamer] creating tx stream " << tx_hints.to_string() << std::endl;
+ UHD_TX_STREAMER_LOG() << "creating tx stream " << tx_hints.to_string() ;
both_xports_t xport = make_transport(stream_address, TX_DATA, tx_hints);
- both_xports_t async_xport = make_transport(stream_address, ASYNC_MSG, device_addr_t(""));
- UHD_STREAMER_LOG() << std::hex << "[TX Streamer] data_sid = " << xport.send_sid << std::dec << std::endl;
+ both_xports_t async_xport = make_transport(stream_address, ASYNC_MSG, device_addr_t(""));
+ UHD_TX_STREAMER_LOG() << std::hex << "[TX Streamer] data_sid = " << xport.send_sid << std::dec << std::endl;
// To calculate the max number of samples per packet, we assume the maximum header length
// to avoid fragmentation should the entire header be used.
const size_t bpp = tx_hints.cast<size_t>("bpp", xport.send->get_send_frame_size()) - stream_options.tx_max_len_hdr;
const size_t bpi = convert::get_bytes_per_item(args.otw_format); // bytes per item
const size_t spp = std::min(args.args.cast<size_t>("spp", bpp/bpi), bpp/bpi); // samples per packet
- UHD_STREAMER_LOG() << "[TX Streamer] spp == " << spp << std::endl;
+ UHD_TX_STREAMER_LOG() << "spp == " << spp ;
//make the new streamer given the samples per packet
if (not my_streamer)
@@ -792,7 +798,7 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
//init some streamer stuff
std::string conv_endianness;
- if (get_transport_endianness(mb_index) == ENDIANNESS_BIG) {
+ if (xport.endianness == ENDIANNESS_BIG) {
my_streamer->set_vrt_packer(&vrt::chdr::if_hdr_pack_be);
conv_endianness = "be";
} else {
@@ -817,7 +823,7 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
tx_hints // This can override the value reported by the block!
);
const size_t fc_handle_window = std::max<size_t>(1, fc_window / stream_options.tx_fc_response_freq);
- UHD_STREAMER_LOG() << "[TX Streamer] Flow Control Window = " << fc_window << ", Flow Control Handler Window = " << fc_handle_window << std::endl;
+ UHD_TX_STREAMER_LOG() << "Flow Control Window = " << fc_window << ", Flow Control Handler Window = " << fc_handle_window ;
blk_ctrl->configure_flow_control_in(
stream_options.tx_fc_response_cycles,
fc_handle_window, /*pkts*/
@@ -841,14 +847,14 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
&handle_tx_async_msgs,
async_tx_info,
my_streamer->_async_xport.recv,
- endianness,
+ xport.endianness,
tick_rate_retriever
)
);
blk_ctrl->sr_write(uhd::rfnoc::SR_CLEAR_RX_FC, 0xc1ea12, block_port);
blk_ctrl->sr_write(uhd::rfnoc::SR_RESP_IN_DST_SID, my_streamer->_async_xport.recv_sid.get_dst(), block_port);
- UHD_STREAMER_LOG() << "[TX Streamer] resp_in_dst_sid == " << boost::format("0x%04X") % xport.recv_sid.get_dst() << std::endl;
+ UHD_TX_STREAMER_LOG() << "resp_in_dst_sid == " << boost::format("0x%04X") % xport.recv_sid.get_dst() ;
// FIXME: Once there is a better way to map the radio block and port
// to the channel or another way to receive asynchronous messages that
@@ -859,8 +865,8 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
uhd::rfnoc::block_id_t radio_id(args.args["radio_id"]);
size_t radio_port = args.args.cast<size_t>("radio_port", 0);
std::vector<boost::shared_ptr<uhd::rfnoc::radio_ctrl> > downstream_radio_nodes = blk_ctrl->find_downstream_node<uhd::rfnoc::radio_ctrl>();
- UHD_STREAMER_LOG() << "[TX Streamer] Number of downstream radio nodes: " << downstream_radio_nodes.size() << std::endl;
- BOOST_FOREACH(const boost::shared_ptr<uhd::rfnoc::radio_ctrl> &node, downstream_radio_nodes) {
+ UHD_TX_STREAMER_LOG() << "Number of downstream radio nodes: " << downstream_radio_nodes.size();
+ for(const boost::shared_ptr<uhd::rfnoc::radio_ctrl> &node: downstream_radio_nodes) {
if (node->get_block_id() == radio_id) {
node->sr_write(uhd::rfnoc::SR_RESP_IN_DST_SID, my_streamer->_async_xport.recv_sid.get_dst(), radio_port);
}
@@ -873,8 +879,8 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
// soon as possible.
// Find all downstream radio nodes and set their response SID to the host
std::vector<boost::shared_ptr<uhd::rfnoc::radio_ctrl> > downstream_radio_nodes = blk_ctrl->find_downstream_node<uhd::rfnoc::radio_ctrl>();
- UHD_STREAMER_LOG() << "[TX Streamer] Number of downstream radio nodes: " << downstream_radio_nodes.size() << std::endl;
- BOOST_FOREACH(const boost::shared_ptr<uhd::rfnoc::radio_ctrl> &node, downstream_radio_nodes) {
+ UHD_TX_STREAMER_LOG() << "Number of downstream radio nodes: " << downstream_radio_nodes.size();
+ for(const boost::shared_ptr<uhd::rfnoc::radio_ctrl> &node: downstream_radio_nodes) {
node->sr_write(uhd::rfnoc::SR_RESP_IN_DST_SID, my_streamer->_async_xport.recv_sid.get_dst(), block_port);
}
}
@@ -884,12 +890,12 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
my_streamer->_xport.send = zero_copy_flow_ctrl::make(
my_streamer->_xport.send,
boost::bind(
- &tx_flow_ctrl,
- fc_cache,
- my_streamer->_xport.recv,
- (endianness == ENDIANNESS_BIG ? uhd::ntohx<uint32_t> : uhd::wtohx<uint32_t>),
- (endianness == ENDIANNESS_BIG ? vrt::chdr::if_hdr_unpack_be : vrt::chdr::if_hdr_unpack_le),
- _1),
+ &tx_flow_ctrl,
+ fc_cache,
+ my_streamer->_xport.recv,
+ (xport.endianness == ENDIANNESS_BIG ? uhd::ntohx<uint32_t> : uhd::wtohx<uint32_t>),
+ (xport.endianness == ENDIANNESS_BIG ? vrt::chdr::if_hdr_unpack_be : vrt::chdr::if_hdr_unpack_le),
+ _1),
NULL);
//Give the streamer a functor to get the send buffer
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-28 19:14:01 +0000