diff options
Diffstat (limited to 'host/lib/include/uhdlib')
| -rw-r--r-- | host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp | 17 | ||||
| -rw-r--r-- | host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp | 175 | ||||
| -rw-r--r-- | host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp | 323 | ||||
| -rw-r--r-- | host/lib/include/uhdlib/rfnoc/rx_flow_ctrl_state.hpp | 17 | ||||
| -rw-r--r-- | host/lib/include/uhdlib/rfnoc/tx_flow_ctrl_state.hpp | 9 |
5 files changed, 133 insertions, 408 deletions
diff --git a/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp b/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp index 9cf2f305f..726ea7f6c 100644 --- a/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp +++ b/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp @@ -7,6 +7,7 @@ #ifndef INCLUDED_RFNOC_CHDR_CTRL_XPORT_HPP #define INCLUDED_RFNOC_CHDR_CTRL_XPORT_HPP +#include <uhdlib/rfnoc/chdr_packet.hpp> #include <uhdlib/rfnoc/chdr_types.hpp> #include <uhdlib/transport/io_service.hpp> #include <mutex> @@ -41,12 +42,18 @@ public: static sptr make(io_service::sptr io_srv, send_link_if::sptr send_link, recv_link_if::sptr recv_link, + const chdr::chdr_packet_factory& pkt_factory, sep_id_t my_epid, size_t num_send_frames, size_t num_recv_frames) { - return std::make_shared<chdr_ctrl_xport>( - io_srv, send_link, recv_link, my_epid, num_send_frames, num_recv_frames); + return std::make_shared<chdr_ctrl_xport>(io_srv, + send_link, + recv_link, + pkt_factory, + my_epid, + num_send_frames, + num_recv_frames); } /*! @@ -62,6 +69,7 @@ public: chdr_ctrl_xport(io_service::sptr io_srv, send_link_if::sptr send_link, recv_link_if::sptr recv_link, + const chdr::chdr_packet_factory& pkt_factory, sep_id_t my_epid, size_t num_send_frames, size_t num_recv_frames); @@ -122,10 +130,15 @@ private: chdr_ctrl_xport(const chdr_ctrl_xport&) = delete; sep_id_t _my_epid; + + // Packet for received data + chdr::chdr_packet::uptr _recv_packet; + send_io_if::sptr _send_if; recv_io_if::sptr _ctrl_recv_if; recv_io_if::sptr _mgmt_recv_if; + // FIXME: Remove this when have threaded_io_service std::mutex _mutex; }; diff --git a/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp b/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp index 41af766b4..9c5d88066 100644 --- a/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp +++ b/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp @@ -8,9 +8,9 @@ #define INCLUDED_LIBUHD_CHDR_RX_DATA_XPORT_HPP #include <uhd/config.hpp> +#include <uhd/exception.hpp> #include <uhdlib/rfnoc/chdr_packet.hpp> #include <uhdlib/rfnoc/chdr_types.hpp> -#include <uhdlib/rfnoc/mgmt_portal.hpp> #include <uhdlib/rfnoc/rfnoc_common.hpp> #include <uhdlib/rfnoc/rx_flow_ctrl_state.hpp> #include <uhdlib/transport/io_service.hpp> @@ -19,6 +19,10 @@ namespace uhd { namespace rfnoc { +namespace mgmt { +class mgmt_portal; +} + namespace detail { /*! @@ -29,21 +33,12 @@ class rx_flow_ctrl_sender public: //! Constructor rx_flow_ctrl_sender( - const chdr::chdr_packet_factory& pkt_factory, const sep_id_pair_t sep_ids) - : _dst_epid(sep_ids.first) - { - _fc_packet = pkt_factory.make_strs(); - _fc_strs_pyld.src_epid = sep_ids.second; - } + const chdr::chdr_packet_factory& pkt_factory, const sep_id_pair_t sep_ids); - /*! Configure buffer capacity + /*! configure buffer capacity * \param recv_capacity The buffer capacity of the receive link */ - void set_capacity(const stream_buff_params_t& recv_capacity) - { - _fc_strs_pyld.capacity_bytes = recv_capacity.bytes; - _fc_strs_pyld.capacity_pkts = recv_capacity.packets; - } + void set_capacity(const stream_buff_params_t& recv_capacity); /*! Send a flow control response packet * @@ -119,113 +114,59 @@ public: const void* payload = nullptr; }; - /*! Constructor + //! Flow control parameters + struct fc_params_t + { + stream_buff_params_t buff_capacity; + stream_buff_params_t freq; + }; + + /*! Configure stream endpoint route and flow control * * \param io_srv The service that will schedule the xport I/O * \param recv_link The recv link, already attached to the I/O service * \param send_link The send link, already attached to the I/O service * \param pkt_factory Factory to create packets with the desired chdr_w and endianness + * \param mgmt_portal Management portal to configure stream endpoint * \param epids Source and destination endpoint IDs * \param pyld_buff_fmt Datatype of SW buffer that holds the data payload * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata - * \param num_recv_frames Num frames to reserve from the recv link * \param recv_capacity Total capacity of the recv link * \param fc_freq Frequency of flow control status messages * \param fc_headroom Headroom for flow control status messages * \param lossy_xport Whether the xport is lossy, for flow control configuration + * \return Parameters for xport flow control */ - chdr_rx_data_xport(uhd::transport::io_service::sptr io_srv, + static fc_params_t configure_sep(uhd::transport::io_service::sptr io_srv, uhd::transport::recv_link_if::sptr recv_link, uhd::transport::send_link_if::sptr send_link, - uhd::rfnoc::mgmt::mgmt_portal& mgmt_portal, const chdr::chdr_packet_factory& pkt_factory, + uhd::rfnoc::mgmt::mgmt_portal& mgmt_portal, const uhd::rfnoc::sep_id_pair_t& epids, const uhd::rfnoc::sw_buff_t pyld_buff_fmt, const uhd::rfnoc::sw_buff_t mdata_buff_fmt, - const size_t num_recv_frames, const stream_buff_params_t& recv_capacity, const stream_buff_params_t& fc_freq, const stream_buff_params_t& fc_headroom, - const bool lossy_xport) - : _fc_state(epids), _fc_sender(pkt_factory, epids), _epid(epids.second) - { - const sep_id_t remote_epid = epids.first; - const sep_id_t local_epid = epids.second; - - UHD_LOG_TRACE("XPORT::RX_DATA_XPORT", - "Creating rx xport with local epid=" << local_epid - << ", remote epid=" << remote_epid); - - _recv_packet = pkt_factory.make_generic(); - _fc_sender.set_capacity(recv_capacity); - - // Calculate max payload size - const size_t pyld_offset = - _recv_packet->calculate_payload_offset(chdr::PKT_TYPE_DATA_WITH_TS); - _max_payload_size = recv_link->get_recv_frame_size() - pyld_offset; - - // Make data transport - auto recv_cb = [this](buff_t::uptr& buff, - transport::recv_link_if* recv_link, - transport::send_link_if* send_link) { - return this->_recv_callback(buff, recv_link, send_link); - }; - - auto fc_cb = [this](buff_t::uptr buff, - transport::recv_link_if* recv_link, - transport::send_link_if* send_link) { - this->_fc_callback(std::move(buff), recv_link, send_link); - }; - - // Needs just a single send frame for responses - _recv_io = io_srv->make_recv_client(recv_link, - num_recv_frames, - recv_cb, - send_link, - /* num_send_frames*/ 1, - fc_cb); - - // Create a control transport with the rx data links to send mgmt packets - // needed to setup the stream - // Piggyback on frames from the recv_io_if - auto ctrl_xport = uhd::rfnoc::chdr_ctrl_xport::make(io_srv, - send_link, - recv_link, - local_epid, - 0, // num_send_frames - 0); // num_recv_frames - - // Setup a route to the EPID - // Note that this may be gratuitous--The endpoint may already have been set up - mgmt_portal.setup_local_route(*ctrl_xport, remote_epid); - - // Initialize flow control - management portal sends a stream command - // containing its requested flow control frequency, the rx transport - // responds with a stream status containing its buffer capacity. - mgmt_portal.config_local_rx_stream_start(*ctrl_xport, - remote_epid, - lossy_xport, - pyld_buff_fmt, - mdata_buff_fmt, - fc_freq, - fc_headroom); - - mgmt_portal.config_local_rx_stream_commit(*ctrl_xport, remote_epid); - - UHD_LOG_TRACE("XPORT::RX_DATA_XPORT", - "Stream endpoint was configured with:" - << std::endl - << "capacity bytes=" << recv_capacity.bytes - << ", packets=" << recv_capacity.packets << std::endl - << "fc headroom bytes=" << fc_headroom.bytes - << ", packets=" << fc_headroom.packets << std::endl - << "fc frequency bytes=" << fc_freq.bytes - << ", packets=" << fc_freq.packets); - - // We no longer need the control xport, release it so - // the control xport is no longer connected to the I/O service. - ctrl_xport.reset(); - } + const bool lossy_xport); + + /*! Constructor + * + * \param io_srv The service that will schedule the xport I/O + * \param recv_link The recv link, already attached to the I/O service + * \param send_link The send link, already attached to the I/O service + * \param pkt_factory Factory to create packets with the desired chdr_w and endianness + * \param epids Source and destination endpoint IDs + * \param num_recv_frames Num frames to reserve from the recv link + * \param fc_params Parameters for flow control + */ + chdr_rx_data_xport(uhd::transport::io_service::sptr io_srv, + uhd::transport::recv_link_if::sptr recv_link, + uhd::transport::send_link_if::sptr send_link, + const chdr::chdr_packet_factory& pkt_factory, + const uhd::rfnoc::sep_id_pair_t& epids, + const size_t num_recv_frames, + const fc_params_t& fc_params); /*! Returns maximum number payload bytes * @@ -286,15 +227,16 @@ private: transport::send_link_if* send_link) { _recv_packet->refresh(buff->data()); - const auto header = _recv_packet->get_chdr_header(); - const auto type = header.get_pkt_type(); - const auto dst_epid = header.get_dst_epid(); - const auto packet_size = buff->packet_size(); + const auto header = _recv_packet->get_chdr_header(); + const auto dst_epid = header.get_dst_epid(); if (dst_epid != _epid) { return false; } + const auto type = header.get_pkt_type(); + const auto packet_size = header.get_length(); + if (type == chdr::PKT_TYPE_STRC) { chdr::strc_payload strc; strc.deserialize(_recv_packet->get_payload_const_ptr_as<uint64_t>(), @@ -316,26 +258,6 @@ private: buff = buff_t::uptr(); _fc_state.xfer_done(packet_size); _send_fc_response(send_link); - } else if (strc.op_code == chdr::STRC_INIT) { - _fc_state.initialize( - {strc.num_bytes, static_cast<uint32_t>(strc.num_pkts)}); - - UHD_LOG_TRACE("XPORT::RX_DATA_XPORT", - "Received strc init with fc freq" - << " bytes=" << strc.num_bytes << ", packets=" << strc.num_pkts); - - // Make sure flow control was initialized - assert(_fc_state.get_fc_freq().bytes > 0); - assert(_fc_state.get_fc_freq().packets > 0); - - // Send a strs response to configure flow control on the sender - _fc_sender.send_strs(send_link, _fc_state.get_xfer_counts()); - - // Reset counts, since mgmt_portal will do it to FPGA - _fc_state.reset_counts(); - - recv_link->release_recv_buff(std::move(buff)); - buff = buff_t::uptr(); } else { throw uhd::value_error("Unexpected opcode value in STRC packet."); } @@ -373,7 +295,9 @@ private: transport::recv_link_if* recv_link, transport::send_link_if* send_link) { - const size_t packet_size = buff->packet_size(); + _recv_packet->refresh(buff->data()); + const auto header = _recv_packet->get_chdr_header(); + const size_t packet_size = header.get_length(); recv_link->release_recv_buff(std::move(buff)); _fc_state.xfer_done(packet_size); _send_fc_response(send_link); @@ -393,8 +317,8 @@ private: } /*! - * Checks if the sequence number is out of sequence, prints 'D' if it is - * and returns result of check. + * Checks if the sequence number is out of sequence, increments sequence + * number for next packet. * * \return true if a sequence error occurred */ @@ -404,7 +328,6 @@ private: _data_seq_num = seq_num + 1; if (expected_packet_count != seq_num) { - UHD_LOG_FASTPATH("D"); return true; } return false; diff --git a/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp b/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp index 63c2b24cb..1e54a2f7a 100644 --- a/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp +++ b/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp @@ -7,10 +7,11 @@ #ifndef INCLUDED_LIBUHD_CHDR_TX_DATA_XPORT_HPP #define INCLUDED_LIBUHD_CHDR_TX_DATA_XPORT_HPP +#include <uhd/exception.hpp> #include <uhd/types/metadata.hpp> +#include <uhd/utils/log.hpp> #include <uhdlib/rfnoc/chdr_packet.hpp> #include <uhdlib/rfnoc/chdr_types.hpp> -#include <uhdlib/rfnoc/mgmt_portal.hpp> #include <uhdlib/rfnoc/rfnoc_common.hpp> #include <uhdlib/rfnoc/tx_flow_ctrl_state.hpp> #include <uhdlib/transport/io_service.hpp> @@ -19,6 +20,10 @@ namespace uhd { namespace rfnoc { +namespace mgmt { +class mgmt_portal; +} + namespace detail { /*! @@ -29,13 +34,7 @@ class tx_flow_ctrl_sender public: //! Constructor tx_flow_ctrl_sender( - const chdr::chdr_packet_factory& pkt_factory, const sep_id_pair_t sep_ids) - : _dst_epid(sep_ids.second) - { - _fc_packet = pkt_factory.make_strc(); - _fc_strc_pyld.src_epid = sep_ids.first; - _fc_strc_pyld.op_code = chdr::STRC_RESYNC; - } + const chdr::chdr_packet_factory& pkt_factory, const sep_id_pair_t sep_ids); /*! * Sends a flow control resync packet @@ -104,9 +103,10 @@ private: class chdr_tx_data_xport { public: - using uptr = std::unique_ptr<chdr_tx_data_xport>; - using buff_t = transport::frame_buff; - using enqueue_async_msg_fn_t = std::function<void(async_metadata_t::event_code_t, bool, uint64_t)>; + using uptr = std::unique_ptr<chdr_tx_data_xport>; + using buff_t = transport::frame_buff; + using enqueue_async_msg_fn_t = + std::function<void(async_metadata_t::event_code_t, bool, uint64_t)>; //! Information about data packet struct packet_info_t @@ -117,84 +117,54 @@ public: size_t payload_bytes = 0; }; - /*! Constructor + //! Flow control parameters + struct fc_params_t + { + stream_buff_params_t buff_capacity; + }; + + /*! Configure route to the sep and flow control * - * \param io_srv The service that will schedule the xport I/O + * \param io_srv The I/O service to be used with the transport * \param recv_link The recv link, already attached to the I/O service * \param send_link The send link, already attached to the I/O service * \param pkt_factory Factory to create packets with the desired chdr_w and endianness + * \param mgmt_portal Management portal to configure stream endpoint * \param epids Source and destination endpoint IDs * \param pyld_buff_fmt Datatype of SW buffer that holds the data payload * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata - * \param num_send_frames Num frames to reserve from the send link * \param fc_freq_ratio Ratio to use to configure the device fc frequency * \param fc_headroom_ratio Ratio to use to configure the device fc headroom + * \return Parameters for xport flow control */ - chdr_tx_data_xport(uhd::transport::io_service::sptr io_srv, + static fc_params_t configure_sep(uhd::transport::io_service::sptr io_srv, uhd::transport::recv_link_if::sptr recv_link, uhd::transport::send_link_if::sptr send_link, - uhd::rfnoc::mgmt::mgmt_portal& mgmt_portal, const chdr::chdr_packet_factory& pkt_factory, + uhd::rfnoc::mgmt::mgmt_portal& mgmt_portal, const uhd::rfnoc::sep_id_pair_t& epids, const uhd::rfnoc::sw_buff_t pyld_buff_fmt, const uhd::rfnoc::sw_buff_t mdata_buff_fmt, - const size_t num_send_frames, const double fc_freq_ratio, - const double fc_headroom_ratio) - : _fc_sender(pkt_factory, epids), _epid(epids.first) - { - const sep_id_t remote_epid = epids.second; - const sep_id_t local_epid = epids.first; - - UHD_LOG_TRACE("XPORT::TX_DATA_XPORT", - "Creating tx xport with local epid=" << local_epid - << ", remote epid=" << remote_epid); - - _send_header.set_dst_epid(epids.second); - _send_packet = pkt_factory.make_generic(); - _recv_packet = pkt_factory.make_generic(); - - // Calculate max payload size - const size_t pyld_offset = - _send_packet->calculate_payload_offset(chdr::PKT_TYPE_DATA_WITH_TS); - _max_payload_size = send_link->get_send_frame_size() - pyld_offset; - - _configure_sep(io_srv, - recv_link, - send_link, - mgmt_portal, - local_epid, - remote_epid, - pyld_buff_fmt, - mdata_buff_fmt); - - _initialize_flow_ctrl(io_srv, - recv_link, - send_link, - pkt_factory, - epids, - fc_freq_ratio, - fc_headroom_ratio); - - // Now create the send I/O we will use for data - auto send_cb = [this](buff_t::uptr& buff, transport::send_link_if* send_link) { - this->_send_callback(buff, send_link); - }; - - auto recv_cb = [this](buff_t::uptr& buff, - transport::recv_link_if* recv_link, - transport::send_link_if* send_link) { - return this->_recv_callback(buff, recv_link, send_link); - }; - - // Needs just a single recv frame for strs packets - _send_io = io_srv->make_send_client(send_link, - num_send_frames, - send_cb, - recv_link, - /* num_recv_frames */ 1, - recv_cb); - } + const double fc_headroom_ratio); + + /*! Constructor + * + * \param io_srv The service that will schedule the xport I/O + * \param recv_link The recv link, already attached to the I/O service + * \param send_link The send link, already attached to the I/O service + * \param pkt_factory Factory to create packets with the desired chdr_w and endianness + * \param epids Source and destination endpoint IDs + * \param num_send_frames Num frames to reserve from the send link + * \param fc_params Parameters for flow control + */ + chdr_tx_data_xport(uhd::transport::io_service::sptr io_srv, + uhd::transport::recv_link_if::sptr recv_link, + uhd::transport::send_link_if::sptr send_link, + const chdr::chdr_packet_factory& pkt_factory, + const uhd::rfnoc::sep_id_pair_t& epids, + const size_t num_send_frames, + const fc_params_t fc_params); /*! Returns maximum number of payload bytes * @@ -300,23 +270,27 @@ private: if (strs.status != chdr::STRS_OKAY) { switch (strs.status) { - case chdr::STRS_SEQERR: - UHD_LOG_FASTPATH("S"); - if (_enqueue_async_msg) { - _enqueue_async_msg(async_metadata_t::EVENT_CODE_SEQ_ERROR, false, 0); - } - break; - case chdr::STRS_DATAERR: - UHD_LOG_WARNING("XPORT::TX_DATA_XPORT", "Received data error in tx stream!"); - break; - case chdr::STRS_RTERR: - UHD_LOG_WARNING("XPORT::TX_DATA_XPORT", "Received routing error in tx stream!"); - break; - case chdr::STRS_CMDERR: - UHD_LOG_WARNING("XPORT::TX_DATA_XPORT", "Received command error in tx stream!"); - break; - default: - break; + case chdr::STRS_SEQERR: + UHD_LOG_FASTPATH("S"); + if (_enqueue_async_msg) { + _enqueue_async_msg( + async_metadata_t::EVENT_CODE_SEQ_ERROR, false, 0); + } + break; + case chdr::STRS_DATAERR: + UHD_LOG_WARNING( + "XPORT::TX_DATA_XPORT", "Received data error in tx stream!"); + break; + case chdr::STRS_RTERR: + UHD_LOG_WARNING("XPORT::TX_DATA_XPORT", + "Received routing error in tx stream!"); + break; + case chdr::STRS_CMDERR: + UHD_LOG_WARNING("XPORT::TX_DATA_XPORT", + "Received command error in tx stream!"); + break; + default: + break; } } @@ -359,177 +333,6 @@ private: } } - /*! - * Configures the stream endpoint using mgmt_portal - */ - void _configure_sep(uhd::transport::io_service::sptr io_srv, - uhd::transport::recv_link_if::sptr recv_link, - uhd::transport::send_link_if::sptr send_link, - uhd::rfnoc::mgmt::mgmt_portal& mgmt_portal, - const uhd::rfnoc::sep_id_t& local_epid, - const uhd::rfnoc::sep_id_t& remote_epid, - const uhd::rfnoc::sw_buff_t pyld_buff_fmt, - const uhd::rfnoc::sw_buff_t mdata_buff_fmt) - { - // Create a control transport with the tx data links to send mgmt packets - // needed to setup the stream. Only need one frame for this. - auto ctrl_xport = uhd::rfnoc::chdr_ctrl_xport::make(io_srv, - send_link, - recv_link, - local_epid, - 1, // num_send_frames - 1); // num_recv_frames - - // Setup a route to the EPID - mgmt_portal.setup_local_route(*ctrl_xport, remote_epid); - - mgmt_portal.config_local_tx_stream( - *ctrl_xport, remote_epid, pyld_buff_fmt, mdata_buff_fmt); - - // We no longer need the control xport, release it so - // the control xport is no longer connected to the I/O service. - ctrl_xport.reset(); - } - - /*! - * Initializes flow control - * - * To initialize flow control, we need to send an init strc packet, then - * receive a strs containing the stream endpoint ingress buffer size. We - * then repeat this (now that we know the buffer size) to configure the flow - * control frequency. To avoid having this logic within the data packet - * processing flow, we use temporary send and recv I/O instances with - * simple callbacks here. - */ - void _initialize_flow_ctrl(uhd::transport::io_service::sptr io_srv, - uhd::transport::recv_link_if::sptr recv_link, - uhd::transport::send_link_if::sptr send_link, - const chdr::chdr_packet_factory& pkt_factory, - const sep_id_pair_t sep_ids, - const double fc_freq_ratio, - const double fc_headroom_ratio) - { - // No flow control at initialization, just release all send buffs - auto send_cb = [this](buff_t::uptr& buff, transport::send_link_if* send_link) { - send_link->release_send_buff(std::move(buff)); - buff = nullptr; - }; - - // For recv, just queue strs packets for recv_io to read - auto recv_cb = [this](buff_t::uptr& buff, - transport::recv_link_if* /*recv_link*/, - transport::send_link_if* /*send_link*/) { - _recv_packet->refresh(buff->data()); - const auto header = _recv_packet->get_chdr_header(); - const auto type = header.get_pkt_type(); - const auto dst_epid = header.get_dst_epid(); - - return (dst_epid == _epid && type == chdr::PKT_TYPE_STRS); - }; - - // No flow control at initialization, just release all recv buffs - auto fc_cb = [this](buff_t::uptr buff, - transport::recv_link_if* recv_link, - transport::send_link_if* /*send_link*/) { - recv_link->release_recv_buff(std::move(buff)); - }; - - auto send_io = io_srv->make_send_client(send_link, - 1, // num_send_frames - send_cb, - nullptr, - 0, // num_recv_frames - nullptr); - - auto recv_io = io_srv->make_recv_client(recv_link, - 1, // num_recv_frames - recv_cb, - nullptr, - 0, // num_send_frames - fc_cb); - - chdr::chdr_strc_packet::uptr strc_packet = pkt_factory.make_strc(); - chdr::chdr_packet::uptr& recv_packet = _recv_packet; - - // Function to send a strc init - auto send_strc_init = [&send_io, sep_ids, &strc_packet]( - const stream_buff_params_t fc_freq = {0, 0}) { - transport::frame_buff::uptr buff = send_io->get_send_buff(0); - - if (!buff) { - throw uhd::runtime_error( - "tx xport timed out getting a send buffer for strc init"); - } - - chdr::chdr_header header; - header.set_seq_num(0); - header.set_dst_epid(sep_ids.second); - - chdr::strc_payload strc_pyld; - strc_pyld.src_epid = sep_ids.first; - strc_pyld.op_code = chdr::STRC_INIT; - strc_pyld.num_bytes = fc_freq.bytes; - strc_pyld.num_pkts = fc_freq.packets; - strc_packet->refresh(buff->data(), header, strc_pyld); - - const size_t size = header.get_length(); - buff->set_packet_size(size); - send_io->release_send_buff(std::move(buff)); - }; - - // Function to receive a strs, returns buffer capacity - auto recv_strs = [&recv_io, &recv_packet]() -> stream_buff_params_t { - transport::frame_buff::uptr buff = recv_io->get_recv_buff(200); - - if (!buff) { - throw uhd::runtime_error( - "tx xport timed out wating for a strs packet during initialization"); - } - - recv_packet->refresh(buff->data()); - UHD_ASSERT_THROW( - recv_packet->get_chdr_header().get_pkt_type() == chdr::PKT_TYPE_STRS); - chdr::strs_payload strs; - strs.deserialize(recv_packet->get_payload_const_ptr_as<uint64_t>(), - recv_packet->get_payload_size() / sizeof(uint64_t), - recv_packet->conv_to_host<uint64_t>()); - - recv_io->release_recv_buff(std::move(buff)); - - return {strs.capacity_bytes, static_cast<uint32_t>(strs.capacity_pkts)}; - }; - - // Send a strc init to get the buffer size - send_strc_init(); - stream_buff_params_t capacity = recv_strs(); - _fc_state.set_dest_capacity(capacity); - - UHD_LOG_TRACE("XPORT::TX_DATA_XPORT", - "Received strs initializing buffer capacity to " << capacity.bytes - << " bytes"); - - // Calculate the requested fc_freq parameters - uhd::rfnoc::stream_buff_params_t fc_freq = { - static_cast<uint64_t>(std::ceil(double(capacity.bytes) * fc_freq_ratio)), - static_cast<uint32_t>(std::ceil(double(capacity.packets) * fc_freq_ratio))}; - - const size_t headroom_bytes = - static_cast<uint64_t>(std::ceil(double(capacity.bytes) * fc_headroom_ratio)); - const size_t headroom_packets = static_cast<uint32_t>( - std::ceil(double(capacity.packets) * fc_headroom_ratio)); - - fc_freq.bytes -= headroom_bytes; - fc_freq.packets -= headroom_packets; - - // Send a strc init to configure fc freq - send_strc_init(fc_freq); - recv_strs(); - - // Release temporary I/O service interfaces to disconnect from it - send_io.reset(); - recv_io.reset(); - } - // Interface to the I/O service transport::send_io_if::sptr _send_io; diff --git a/host/lib/include/uhdlib/rfnoc/rx_flow_ctrl_state.hpp b/host/lib/include/uhdlib/rfnoc/rx_flow_ctrl_state.hpp index 937baf982..ed6553bf3 100644 --- a/host/lib/include/uhdlib/rfnoc/rx_flow_ctrl_state.hpp +++ b/host/lib/include/uhdlib/rfnoc/rx_flow_ctrl_state.hpp @@ -17,12 +17,10 @@ class rx_flow_ctrl_state { public: //! Constructor - rx_flow_ctrl_state(const rfnoc::sep_id_pair_t epids) : _epids(epids) {} - - //! Initialize frequency parameters - void initialize(const stream_buff_params_t fc_freq) + rx_flow_ctrl_state( + const rfnoc::sep_id_pair_t epids, const stream_buff_params_t fc_freq) + : _fc_freq(fc_freq), _epids(epids) { - _fc_freq = fc_freq; } //! Resynchronize with transfer counts from the sender @@ -50,15 +48,6 @@ public: } } - //! Reset the transfer counts (happens during init) - void reset_counts() - { - UHD_LOGGER_TRACE("rx_flow_ctrl_state") - << "Resetting transfer counts" << std::endl; - _recv_counts = {0, 0}; - _xfer_counts = {0, 0}; - } - //! Update state when data is received void data_received(const size_t bytes) { diff --git a/host/lib/include/uhdlib/rfnoc/tx_flow_ctrl_state.hpp b/host/lib/include/uhdlib/rfnoc/tx_flow_ctrl_state.hpp index 65fc1b093..0005e7584 100644 --- a/host/lib/include/uhdlib/rfnoc/tx_flow_ctrl_state.hpp +++ b/host/lib/include/uhdlib/rfnoc/tx_flow_ctrl_state.hpp @@ -15,11 +15,8 @@ namespace uhd { namespace rfnoc { class tx_flow_ctrl_state { public: - //! Updates destination capacity - void set_dest_capacity(const stream_buff_params_t& capacity) - { - _dest_capacity = capacity; - } + //! Constructor + tx_flow_ctrl_state(const stream_buff_params_t& capacity) : _dest_capacity(capacity) {} //! Updates destination received count void update_dest_recv_count(const stream_buff_params_t& recv_count) @@ -67,7 +64,7 @@ public: //! Clears fc resync request pending status void clear_fc_resync_req_pending() { - _fc_resync_req = false; + _fc_resync_req = false; _last_fc_resync_bytes = _xfer_counts.bytes; } |