diff options
| -rw-r--r-- | host/include/uhd/transport/alignment_buffer.ipp | 8 | ||||
| -rw-r--r-- | host/include/uhd/transport/bounded_buffer.ipp | 39 | ||||
| -rw-r--r-- | host/lib/transport/udp_zero_copy_asio.cpp | 114 |
3 files changed, 99 insertions, 62 deletions
diff --git a/host/include/uhd/transport/alignment_buffer.ipp b/host/include/uhd/transport/alignment_buffer.ipp index 5f09de0d9..833b5d399 100644 --- a/host/include/uhd/transport/alignment_buffer.ipp +++ b/host/include/uhd/transport/alignment_buffer.ipp @@ -54,14 +54,14 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE bool pop_elems_with_timed_wait( std::vector<elem_type> &elems, double timeout ){ - boost::system_time exit_time = boost::get_system_time() + boost::posix_time::microseconds(long(timeout*1e6)); + boost::system_time exit_time = boost::get_system_time() + to_time_dur(timeout); buff_contents_type buff_contents_tmp; std::list<size_t> indexes_to_do(_all_indexes); //do an initial pop to load an initial sequence id size_t index = indexes_to_do.front(); if (not _buffs[index]->pop_with_timed_wait( - buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() + buff_contents_tmp, from_time_dur(exit_time - boost::get_system_time()) )) return false; elems[index] = buff_contents_tmp.first; seq_type expected_seq_id = buff_contents_tmp.second; @@ -76,7 +76,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ indexes_to_do = _all_indexes; index = indexes_to_do.front(); if (not _buffs[index]->pop_with_timed_wait( - buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() + buff_contents_tmp, from_time_dur(exit_time - boost::get_system_time()) )) return false; elems[index] = buff_contents_tmp.first; expected_seq_id = buff_contents_tmp.second; @@ -86,7 +86,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ //pop an element off for this index index = indexes_to_do.front(); if (not _buffs[index]->pop_with_timed_wait( - buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() + buff_contents_tmp, from_time_dur(exit_time - boost::get_system_time()) )) return false; //if the sequence id matches: diff --git a/host/include/uhd/transport/bounded_buffer.ipp b/host/include/uhd/transport/bounded_buffer.ipp index 58f78bab4..edc7faa06 100644 --- a/host/include/uhd/transport/bounded_buffer.ipp +++ b/host/include/uhd/transport/bounded_buffer.ipp @@ -19,18 +19,28 @@ #define INCLUDED_UHD_TRANSPORT_BOUNDED_BUFFER_IPP #include <boost/bind.hpp> +#include <boost/function.hpp> #include <boost/circular_buffer.hpp> #include <boost/thread/condition.hpp> #include <boost/date_time/posix_time/posix_time_types.hpp> namespace uhd{ namespace transport{ namespace{ /*anon*/ + 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(); + } + template <typename elem_type> class bounded_buffer_impl : public bounded_buffer<elem_type>{ public: bounded_buffer_impl(size_t capacity) : _buffer(capacity){ - /* NOP */ + _not_full_fcn = boost::bind(&bounded_buffer_impl<elem_type>::not_full, this); + _not_empty_fcn = boost::bind(&bounded_buffer_impl<elem_type>::not_empty, this); } UHD_INLINE bool push_with_pop_on_full(const elem_type &elem){ @@ -52,17 +62,16 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE void push_with_wait(const elem_type &elem){ boost::unique_lock<boost::mutex> lock(_mutex); - _full_cond.wait(lock, boost::bind(&bounded_buffer_impl<elem_type>::not_full, this)); + _full_cond.wait(lock, _not_full_fcn); _buffer.push_front(elem); lock.unlock(); _empty_cond.notify_one(); } - bool push_with_timed_wait(const elem_type &elem, double timeout){ + UHD_INLINE bool push_with_timed_wait(const elem_type &elem, double timeout){ boost::unique_lock<boost::mutex> lock(_mutex); if (not _full_cond.timed_wait( - lock, boost::posix_time::microseconds(long(timeout*1e6)), - boost::bind(&bounded_buffer_impl<elem_type>::not_full, this) + lock, to_time_dur(timeout), _not_full_fcn )) return false; _buffer.push_front(elem); lock.unlock(); @@ -72,19 +81,18 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE void pop_with_wait(elem_type &elem){ boost::unique_lock<boost::mutex> lock(_mutex); - _empty_cond.wait(lock, boost::bind(&bounded_buffer_impl<elem_type>::not_empty, this)); - this->pop_back(elem); + _empty_cond.wait(lock, _not_empty_fcn); + elem = this->pop_back(); lock.unlock(); _full_cond.notify_one(); } - bool pop_with_timed_wait(elem_type &elem, double timeout){ + UHD_INLINE bool pop_with_timed_wait(elem_type &elem, double timeout){ boost::unique_lock<boost::mutex> lock(_mutex); if (not _empty_cond.timed_wait( - lock, boost::posix_time::microseconds(long(timeout*1e6)), - boost::bind(&bounded_buffer_impl<elem_type>::not_empty, this) + lock, to_time_dur(timeout), _not_empty_fcn )) return false; - this->pop_back(elem); + elem = this->pop_back(); lock.unlock(); _full_cond.notify_one(); return true; @@ -92,7 +100,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE void clear(void){ boost::unique_lock<boost::mutex> lock(_mutex); - while (not_empty()) _buffer.pop_back(); + while (not_empty()) this->pop_back(); lock.unlock(); _full_cond.notify_one(); } @@ -105,16 +113,19 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ bool not_full(void) const{return not _buffer.full();} bool not_empty(void) const{return not _buffer.empty();} + boost::function<bool(void)> _not_full_fcn, _not_empty_fcn; + /*! * Three part operation to pop an element: * 1) assign elem to the back element * 2) assign the back element to empty * 3) pop the back to move the counter */ - UHD_INLINE void pop_back(elem_type &elem){ - elem = _buffer.back(); + UHD_INLINE elem_type pop_back(void){ + elem_type elem = _buffer.back(); _buffer.back() = elem_type(); _buffer.pop_back(); + return elem; } }; }}} //namespace diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 40fe3fa0f..d84aeefdd 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -39,6 +39,10 @@ namespace asio = boost::asio; //Otherwise, get_recv_buff uses a blocking receive with timeout. //#define USE_ASIO_ASYNC_RECV +//Define this to the the boost async io calls to perform send. +//Otherwise, the commit callback uses a blocking send. +//#define USE_ASIO_ASYNC_SEND + //enough buffering for half a second of samples at full rate on usrp2 static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(4 * 25e6 * 0.5); @@ -55,7 +59,13 @@ static const size_t DEFAULT_NUM_RECV_FRAMES = 32; #else static const size_t DEFAULT_NUM_RECV_FRAMES = MIN_RECV_SOCK_BUFF_SIZE/udp_simple::mtu; #endif +//The non-async send only ever requires a single frame +//because the buffer will be committed before a new get. +#ifdef USE_ASIO_ASYNC_SEND static const size_t DEFAULT_NUM_SEND_FRAMES = 32; +#else +static const size_t DEFAULT_NUM_SEND_FRAMES = MIN_SEND_SOCK_BUFF_SIZE/udp_simple::mtu;; +#endif //a single concurrent thread for io_service seems to be the fastest static const size_t CONCURRENCY_HINT = 1; @@ -143,6 +153,12 @@ public: return get_buff_size<Opt>(); } + //! handle a recv callback -> push the filled memory into the fifo + UHD_INLINE void handle_recv(void *mem, size_t len){ + boost::this_thread::disable_interruption di; //disable because the wait can throw + _pending_recv_buffs->push_with_wait(boost::asio::buffer(mem, len)); + } + //////////////////////////////////////////////////////////////////// #ifdef USE_ASIO_ASYNC_RECV //////////////////////////////////////////////////////////////////// @@ -162,16 +178,10 @@ public: return managed_recv_buffer::sptr(); } - //! handle a recv callback -> push the filled memory into the fifo - void handle_recv(void *mem, size_t len){ - boost::this_thread::disable_interruption di; //disable because the wait can throw - _pending_recv_buffs->push_with_wait(boost::asio::buffer(mem, len)); - } - //! release a recv buffer -> start an async recv on the buffer void release(void *mem){ _socket->async_receive( - boost::asio::buffer(mem, _recv_frame_size), + boost::asio::buffer(mem, this->get_recv_frame_size()), boost::bind( &udp_zero_copy_asio_impl::handle_recv, shared_from_this(), mem, @@ -185,6 +195,7 @@ public: //////////////////////////////////////////////////////////////////// managed_recv_buffer::sptr get_recv_buff(double timeout){ boost::this_thread::disable_interruption di; //disable because the wait can throw + asio::mutable_buffer buff; //setup timeval for timeout timeval tv; @@ -196,34 +207,30 @@ public: FD_ZERO(&rset); FD_SET(_sock_fd, &rset); - //call select to perform timed wait - if (::select(_sock_fd+1, &rset, NULL, NULL, &tv) <= 0) - return managed_recv_buffer::sptr(); - - //grab an available buffer - asio::mutable_buffer buff; - if (not _pending_recv_buffs->pop_with_timed_wait(buff, timeout)) - return managed_recv_buffer::sptr(); - - //receive and return the buffer - return managed_recv_buffer::make_safe( - asio::buffer( - boost::asio::buffer_cast<void *>(buff), - _socket->receive(boost::asio::buffer(buff)) - ), - boost::bind( - &udp_zero_copy_asio_impl::release, - shared_from_this(), - asio::buffer_cast<void*>(buff) - ) - ); + //call select to perform timed wait and grab an available buffer with wait + //if the condition is true, call receive and return the managed buffer + if ( + ::select(_sock_fd+1, &rset, NULL, NULL, &tv) > 0 and + _pending_recv_buffs->pop_with_timed_wait(buff, timeout) + ){ + return managed_recv_buffer::make_safe( + asio::buffer( + boost::asio::buffer_cast<void *>(buff), + _socket->receive(asio::buffer(buff)) + ), + boost::bind( + &udp_zero_copy_asio_impl::release, + shared_from_this(), + asio::buffer_cast<void*>(buff) + ) + ); + } + return managed_recv_buffer::sptr(); } void release(void *mem){ boost::this_thread::disable_interruption di; //disable because the wait can throw - _pending_recv_buffs->push_with_wait( - boost::asio::buffer(mem, this->get_recv_frame_size()) - ); + handle_recv(mem, this->get_recv_frame_size()); } //////////////////////////////////////////////////////////////////// @@ -233,6 +240,12 @@ public: size_t get_num_recv_frames(void) const {return _num_recv_frames;} size_t get_recv_frame_size(void) const {return _recv_frame_size;} + //! handle a send callback -> push the emptied memory into the fifo + UHD_INLINE void handle_send(void *mem){ + boost::this_thread::disable_interruption di; //disable because the wait can throw + _pending_send_buffs->push_with_wait(boost::asio::buffer(mem, this->get_send_frame_size())); + } + //! pop an empty send buffer off of the fifo and bind with the commit callback managed_send_buffer::sptr get_send_buff(double timeout){ boost::this_thread::disable_interruption di; //disable because the wait can throw @@ -249,12 +262,9 @@ public: return managed_send_buffer::sptr(); } - //! handle a send callback -> push the emptied memory into the fifo - void handle_send(void *mem){ - boost::this_thread::disable_interruption di; //disable because the wait can throw - _pending_send_buffs->push_with_wait(boost::asio::buffer(mem, _send_frame_size)); - } - + //////////////////////////////////////////////////////////////////// + #ifdef USE_ASIO_ASYNC_SEND + //////////////////////////////////////////////////////////////////// //! commit a send buffer -> start an async send on the buffer void commit(void *mem, size_t len){ _socket->async_send( @@ -266,6 +276,18 @@ public: ); } + //////////////////////////////////////////////////////////////////// + #else /*USE_ASIO_ASYNC_SEND*/ + //////////////////////////////////////////////////////////////////// + void commit(void *mem, size_t len){ + _socket->send(asio::buffer(mem, len)); + handle_send(mem); + } + + //////////////////////////////////////////////////////////////////// + #endif /*USE_ASIO_ASYNC_SEND*/ + //////////////////////////////////////////////////////////////////// + size_t get_num_send_frames(void) const {return _num_send_frames;} size_t get_send_frame_size(void) const {return _send_frame_size;} @@ -297,6 +319,13 @@ template<typename Opt> static void resize_buff_helper( if (name == "recv") min_sock_buff_size = MIN_RECV_SOCK_BUFF_SIZE; if (name == "send") min_sock_buff_size = MIN_SEND_SOCK_BUFF_SIZE; + std::string help_message; + #if defined(UHD_PLATFORM_LINUX) + help_message = str(boost::format( + "Please run: sudo sysctl -w net.core.%smem_max=%d\n" + ) % ((name == "recv")?"r":"w") % min_sock_buff_size); + #endif /*defined(UHD_PLATFORM_LINUX)*/ + //resize the buffer if size was provided if (target_size > 0){ size_t actual_size = udp_trans->resize_buff<Opt>(target_size); @@ -308,13 +337,10 @@ template<typename Opt> static void resize_buff_helper( "Current %s sock buff size: %d bytes" ) % name % actual_size << std::endl; if (actual_size < target_size) uhd::print_warning(str(boost::format( - "The %1% buffer is smaller than the requested size.\n" - "The minimum recommended buffer size is %2% bytes.\n" - "See the transport application notes on buffer resizing.\n" - #if defined(UHD_PLATFORM_LINUX) - "Please run: sudo sysctl -w net.core.rmem_max=%2%\n" - #endif /*defined(UHD_PLATFORM_LINUX)*/ - ) % name % min_sock_buff_size)); + "The %s buffer is smaller than the requested size.\n" + "The minimum recommended buffer size is %d bytes.\n" + "See the transport application notes on buffer resizing.\n%s" + ) % name % min_sock_buff_size % help_message)); } //only enable on platforms that are happy with the large buffer resize |