diff options
Diffstat (limited to 'host/lib/usrp/usrp2/io_impl.cpp')
| -rw-r--r-- | host/lib/usrp/usrp2/io_impl.cpp | 139 |
1 files changed, 77 insertions, 62 deletions
diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp index 30eaecae2..b20b6652e 100644 --- a/host/lib/usrp/usrp2/io_impl.cpp +++ b/host/lib/usrp/usrp2/io_impl.cpp @@ -24,14 +24,24 @@ #include <boost/format.hpp> #include <boost/bind.hpp> #include <boost/thread.hpp> -#include <boost/date_time/posix_time/posix_time_types.hpp> #include <iostream> -#include <list> 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)); +} + +static UHD_INLINE double from_time_dur(const pt::time_duration &time_dur){ + return 1e-6*time_dur.total_microseconds(); +} /*********************************************************************** * constants @@ -61,6 +71,7 @@ public: _last_seq_out = 0; _last_seq_ack = 0; _max_seqs_out = max_seqs_out; + _ready_fcn = boost::bind(&flow_control_monitor::ready, this); } /*! @@ -73,11 +84,8 @@ public: boost::this_thread::disable_interruption di; //disable because the wait can throw boost::unique_lock<boost::mutex> lock(_fc_mutex); _last_seq_out = seq; - return _fc_cond.timed_wait( - lock, - boost::posix_time::microseconds(long(timeout*1e6)), - boost::bind(&flow_control_monitor::ready, this) - ); + if (this->ready()) return true; + return _fc_cond.timed_wait(lock, to_time_dur(timeout), _ready_fcn); } /*! @@ -99,6 +107,7 @@ private: boost::mutex _fc_mutex; boost::condition _fc_cond; seq_type _last_seq_out, _last_seq_ack, _max_seqs_out; + boost::function<bool(void)> _ready_fcn; }; /*********************************************************************** @@ -110,11 +119,15 @@ private: **********************************************************************/ struct usrp2_impl::io_impl{ - io_impl(size_t send_frame_size, size_t width): - packet_handler_recv_state(width), - async_msg_fifo(bounded_buffer<async_metadata_t>::make(100/*messages deep*/)) + io_impl(size_t send_frame_size, const std::vector<zero_copy_if::sptr> &xports): + xports(xports), + get_recv_buffs_fcn(boost::bind(&usrp2_impl::io_impl::get_recv_buffs, this, _1)), + get_send_buffs_fcn(boost::bind(&usrp2_impl::io_impl::get_send_buffs, this, _1)), + packet_handler_recv_state(xports.size()), + packet_handler_send_state(xports.size()), + async_msg_fifo(100/*messages deep*/) { - for (size_t i = 0; i < width; i++){ + for (size_t i = 0; i < xports.size(); i++){ fc_mons.push_back(flow_control_monitor::sptr( new flow_control_monitor(usrp2_impl::sram_bytes/send_frame_size) )); @@ -135,31 +148,32 @@ struct usrp2_impl::io_impl{ recv_pirate_crew.join_all(); } - bool get_send_buffs( - const std::vector<zero_copy_if::sptr> &trans, - vrt_packet_handler::managed_send_buffs_t &buffs, - double timeout - ){ - UHD_ASSERT_THROW(trans.size() == buffs.size()); + bool get_send_buffs(vrt_packet_handler::managed_send_buffs_t &buffs){ + UHD_ASSERT_THROW(xports.size() == buffs.size()); //calculate the flow control word const boost::uint32_t fc_word32 = packet_handler_send_state.next_packet_seq; //grab a managed buffer for each index for (size_t i = 0; i < buffs.size(); i++){ - if (not fc_mons[i]->check_fc_condition(fc_word32, timeout)) return false; - buffs[i] = trans[i]->get_send_buff(timeout); + if (not fc_mons[i]->check_fc_condition(fc_word32, send_timeout)) return false; + buffs[i] = xports[i]->get_send_buff(send_timeout); if (not buffs[i].get()) return false; buffs[i]->cast<boost::uint32_t *>()[0] = uhd::htonx(fc_word32); } return true; } - bool get_recv_buffs( - const std::vector<zero_copy_if::sptr> &xports, - vrt_packet_handler::managed_recv_buffs_t &buffs, - double timeout - ); + bool get_recv_buffs(vrt_packet_handler::managed_recv_buffs_t &buffs); + + const std::vector<zero_copy_if::sptr> &xports; + + //timeouts set on calls to recv/send (passed into get buffs methods) + double recv_timeout, send_timeout; + + //bound callbacks for get buffs (bound once here, not in fast-path) + vrt_packet_handler::get_recv_buffs_t get_recv_buffs_fcn; + vrt_packet_handler::get_send_buffs_t get_send_buffs_fcn; //previous state for each buffer std::vector<vrt::if_packet_info_t> prev_infos; @@ -175,7 +189,7 @@ struct usrp2_impl::io_impl{ void recv_pirate_loop(zero_copy_if::sptr, usrp2_mboard_impl::sptr, size_t); boost::thread_group recv_pirate_crew; bool recv_pirate_crew_raiding; - bounded_buffer<async_metadata_t>::sptr async_msg_fifo; + bounded_buffer<async_metadata_t> async_msg_fifo; boost::mutex spawn_mutex; }; @@ -228,7 +242,7 @@ void usrp2_impl::io_impl::recv_pirate_loop( //print the famous U, and push the metadata into the message queue if (metadata.event_code & underflow_flags) std::cerr << "U" << std::flush; //else std::cout << "metadata.event_code " << metadata.event_code << std::endl; - async_msg_fifo->push_with_pop_on_full(metadata); + async_msg_fifo.push_with_pop_on_full(metadata); } else{ //TODO unknown received packet, may want to print error... @@ -248,7 +262,7 @@ void usrp2_impl::io_init(void){ const size_t send_frame_size = _data_transports.front()->get_send_frame_size(); //create new io impl - _io_impl = UHD_PIMPL_MAKE(io_impl, (send_frame_size, _data_transports.size())); + _io_impl = UHD_PIMPL_MAKE(io_impl, (send_frame_size, _data_transports)); //create a new pirate thread for each zc if (yarr!!) for (size_t i = 0; i < _data_transports.size(); i++){ @@ -274,7 +288,7 @@ 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); + return _io_impl->async_msg_fifo.pop_with_timed_wait(async_metadata, timeout); } /*********************************************************************** @@ -291,10 +305,11 @@ size_t usrp2_impl::get_max_send_samps_per_packet(void) const{ } size_t usrp2_impl::send( - const std::vector<const void *> &buffs, size_t num_samps, + const send_buffs_type &buffs, size_t num_samps, const tx_metadata_t &metadata, const io_type_t &io_type, send_mode_t send_mode, double timeout ){ + _io_impl->send_timeout = timeout; return vrt_packet_handler::send( _io_impl->packet_handler_send_state, //last state of the send handler buffs, num_samps, //buffer to fill @@ -302,7 +317,7 @@ size_t usrp2_impl::send( io_type, _tx_otw_type, //input and output types to convert _mboards.front()->get_master_clock_freq(), //master clock tick rate uhd::transport::vrt::if_hdr_pack_be, - boost::bind(&usrp2_impl::io_impl::get_send_buffs, _io_impl.get(), _data_transports, _1, timeout), + _io_impl->get_send_buffs_fcn, get_max_send_samps_per_packet(), vrt_send_header_offset_words32 ); @@ -311,14 +326,6 @@ size_t usrp2_impl::send( /*********************************************************************** * Alignment logic on receive **********************************************************************/ -static UHD_INLINE boost::posix_time::time_duration to_time_dur(double timeout){ - return boost::posix_time::microseconds(long(timeout*1e6)); -} - -static UHD_INLINE double from_time_dur(const boost::posix_time::time_duration &time_dur){ - return 1e-6*time_dur.total_microseconds(); -} - static UHD_INLINE time_spec_t extract_time_spec( const vrt::if_packet_info_t &packet_info ){ @@ -359,13 +366,24 @@ static UHD_INLINE bool handle_msg_packet( return true; } +class alignment_indexes{ +public: + void reset(size_t len){_indexes = (1 << len) - 1;} + size_t front(void){ //TODO replace with look-up table + size_t index = 0; + while ((_indexes & (1 << index)) == 0) index++; + return index; + } + void remove(size_t index){_indexes &= ~(1 << index);} + bool empty(void){return _indexes == 0;} +private: size_t _indexes; +}; + UHD_INLINE bool usrp2_impl::io_impl::get_recv_buffs( - const std::vector<zero_copy_if::sptr> &xports, - vrt_packet_handler::managed_recv_buffs_t &buffs, - double timeout + vrt_packet_handler::managed_recv_buffs_t &buffs ){ if (buffs.size() == 1){ - buffs[0] = xports[0]->get_recv_buff(timeout); + buffs[0] = xports[0]->get_recv_buff(recv_timeout); if (buffs[0].get() == NULL) return false; bool clear, msg; time_spec_t time; //unused variables //call extract_packet_info to handle printing the overflows @@ -373,16 +391,15 @@ UHD_INLINE bool usrp2_impl::io_impl::get_recv_buffs( return true; } //-------------------- begin alignment logic ---------------------// - boost::system_time exit_time = boost::get_system_time() + to_time_dur(timeout); + boost::system_time exit_time = boost::get_system_time() + to_time_dur(recv_timeout); managed_recv_buffer::sptr buff_tmp; - std::list<size_t> _all_indexes, indexes_to_do; - for (size_t i = 0; i < buffs.size(); i++) _all_indexes.push_back(i); + alignment_indexes indexes_to_do; bool clear, msg; time_spec_t expected_time; //respond to a clear by starting from scratch got_clear: - indexes_to_do = _all_indexes; + indexes_to_do.reset(buffs.size()); clear = false; //do an initial pop to load an initial sequence id @@ -393,10 +410,10 @@ UHD_INLINE bool usrp2_impl::io_impl::get_recv_buffs( if (clear) goto got_clear; buffs[index] = buff_tmp; if (msg) return handle_msg_packet(buffs, index); - indexes_to_do.pop_front(); + indexes_to_do.remove(index); //get an aligned set of elements from the buffers: - while(indexes_to_do.size() != 0){ + while(not indexes_to_do.empty()){ //pop an element off for this index index = indexes_to_do.front(); @@ -411,25 +428,22 @@ UHD_INLINE bool usrp2_impl::io_impl::get_recv_buffs( //if the sequence id matches: // remove this index from the list and continue if (this_time == expected_time){ - indexes_to_do.pop_front(); - continue; - } - - //if the sequence id is older: - // continue with the same index to try again - else if (this_time < expected_time){ - continue; + indexes_to_do.remove(index); } //if the sequence id is newer: // use the new expected time for comparison // add all other indexes back into the list - else{ + else if (this_time > expected_time){ expected_time = this_time; - indexes_to_do = _all_indexes; + indexes_to_do.reset(buffs.size()); indexes_to_do.remove(index); - continue; } + + //if the sequence id is older: + // continue with the same index to try again + //else if (this_time < expected_time)... + } return true; //-------------------- end alignment logic -----------------------// @@ -454,10 +468,11 @@ static void handle_overflow(std::vector<usrp2_mboard_impl::sptr> &mboards, size_ } size_t usrp2_impl::recv( - const std::vector<void *> &buffs, size_t num_samps, + const recv_buffs_type &buffs, size_t num_samps, rx_metadata_t &metadata, const io_type_t &io_type, recv_mode_t recv_mode, double timeout ){ + _io_impl->recv_timeout = timeout; return vrt_packet_handler::recv( _io_impl->packet_handler_recv_state, //last state of the recv handler buffs, num_samps, //buffer to fill @@ -465,7 +480,7 @@ size_t usrp2_impl::recv( io_type, _rx_otw_type, //input and output types to convert _mboards.front()->get_master_clock_freq(), //master clock tick rate uhd::transport::vrt::if_hdr_unpack_be, - boost::bind(&usrp2_impl::io_impl::get_recv_buffs, _io_impl.get(), _data_transports, _1, timeout), - boost::bind(&handle_overflow, _mboards, _1) + _io_impl->get_recv_buffs_fcn, + boost::bind(&handle_overflow, boost::ref(_mboards), _1) ); } |