diff options
| -rw-r--r-- | host/include/uhd/transport/udp_zero_copy.hpp | 2 | ||||
| -rw-r--r-- | host/include/uhd/transport/zero_copy.hpp | 228 | ||||
| -rw-r--r-- | host/lib/transport/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | host/lib/transport/udp_zero_copy_asio.cpp | 99 | ||||
| -rw-r--r-- | host/lib/transport/zero_copy.cpp | 140 |
5 files changed, 303 insertions, 167 deletions
diff --git a/host/include/uhd/transport/udp_zero_copy.hpp b/host/include/uhd/transport/udp_zero_copy.hpp index 525606a9f..818709973 100644 --- a/host/include/uhd/transport/udp_zero_copy.hpp +++ b/host/include/uhd/transport/udp_zero_copy.hpp @@ -34,7 +34,7 @@ namespace uhd{ namespace transport{ * If no platform specific solution is available, make returns a boost asio * implementation that wraps the functionality around a standard send/recv calls. */ -class UHD_API udp_zero_copy : public zero_copy_if{ +class UHD_API udp_zero_copy : public virtual zero_copy_if{ public: typedef boost::shared_ptr<udp_zero_copy> sptr; diff --git a/host/include/uhd/transport/zero_copy.hpp b/host/include/uhd/transport/zero_copy.hpp index 52c6d4143..2efabaccf 100644 --- a/host/include/uhd/transport/zero_copy.hpp +++ b/host/include/uhd/transport/zero_copy.hpp @@ -25,109 +25,169 @@ namespace uhd{ namespace transport{ -/*! - * A managed receive buffer: - * Contains a reference to transport-managed memory, - * and a method to release the memory after reading. - */ -class UHD_API managed_recv_buffer : boost::noncopyable{ -public: - typedef boost::shared_ptr<managed_recv_buffer> sptr; - /*! - * Managed recv buffer destructor: - * Signal to the transport that we are done with the buffer. - * This should be called to release the buffer to the transport. - * After calling, the referenced memory should be considered invalid. + * A managed receive buffer: + * Contains a reference to transport-managed memory, + * and a method to release the memory after reading. */ - virtual ~managed_recv_buffer(void){}; + class UHD_API managed_recv_buffer : boost::noncopyable{ + public: + typedef boost::shared_ptr<managed_recv_buffer> sptr; + + /*! + * Managed recv buffer destructor: + * Signal to the transport that we are done with the buffer. + * This should be called to release the buffer to the transport. + * After calling, the referenced memory should be considered invalid. + */ + virtual ~managed_recv_buffer(void) = 0; + + /*! + * Get the size of the underlying buffer. + * \return the number of bytes + */ + size_t size(void) const{ + return boost::asio::buffer_size(this->get()); + } + + /*! + * Get a pointer to the underlying buffer. + * \return a pointer into memory + */ + template <class T> T cast(void) const{ + return boost::asio::buffer_cast<T>(this->get()); + } + + private: + /*! + * Get a reference to the internal const buffer. + * The buffer has a reference to memory and a size. + * \return a boost asio const buffer + */ + virtual const boost::asio::const_buffer &get(void) const = 0; + }; /*! - * Get the size of the underlying buffer. - * \return the number of bytes + * A managed send buffer: + * Contains a reference to transport-managed memory, + * and a method to release the memory after writing. */ - size_t size(void) const{ - return boost::asio::buffer_size(this->get()); - } + class UHD_API managed_send_buffer : boost::noncopyable{ + public: + typedef boost::shared_ptr<managed_send_buffer> sptr; + + /*! + * Signal to the transport that we are done with the buffer. + * This should be called to commit the write to the transport object. + * After calling, the referenced memory should be considered invalid. + * \param num_bytes the number of bytes written into the buffer + */ + virtual void commit(size_t num_bytes) = 0; + + /*! + * Get the size of the underlying buffer. + * \return the number of bytes + */ + size_t size(void) const{ + return boost::asio::buffer_size(this->get()); + } + + /*! + * Get a pointer to the underlying buffer. + * \return a pointer into memory + */ + template <class T> T cast(void) const{ + return boost::asio::buffer_cast<T>(this->get()); + } + + private: + /*! + * Get a reference to the internal mutable buffer. + * The buffer has a reference to memory and a size. + * \return a boost asio mutable buffer + */ + virtual const boost::asio::mutable_buffer &get(void) const = 0; + }; /*! - * Get a pointer to the underlying buffer. - * \return a pointer into memory + * A zero-copy interface for transport objects. + * Provides a way to get send and receive buffers + * with memory managed by the transport object. */ - template <class T> T cast(void) const{ - return boost::asio::buffer_cast<T>(this->get()); - } + class UHD_API zero_copy_if : boost::noncopyable{ + public: + typedef boost::shared_ptr<zero_copy_if> sptr; -private: - /*! - * Get a reference to the internal const buffer. - * The buffer has a reference to memory and a size. - * \return a boost asio const buffer - */ - virtual const boost::asio::const_buffer &get(void) const = 0; -}; - -/*! - * A managed send buffer: - * Contains a reference to transport-managed memory, - * and a method to release the memory after writing. - */ -class UHD_API managed_send_buffer : boost::noncopyable{ -public: - typedef boost::shared_ptr<managed_send_buffer> sptr; + /*! + * Get a new receive buffer from this transport object. + */ + virtual managed_recv_buffer::sptr get_recv_buff(void) = 0; - /*! - * Signal to the transport that we are done with the buffer. - * This should be called to commit the write to the transport object. - * After calling, the referenced memory should be considered invalid. - * \param num_bytes the number of bytes written into the buffer - */ - virtual void commit(size_t num_bytes) = 0; + /*! + * Get a new send buffer from this transport object. + */ + virtual managed_send_buffer::sptr get_send_buff(void) = 0; + }; /*! - * Get the size of the underlying buffer. - * \return the number of bytes + * A phony-zero-copy interface for transport objects that + * provides a zero-copy interface on top of copying transport. + * This interface implements the get managed recv buffer, + * the base class must implement the private recv method. */ - size_t size(void) const{ - return boost::asio::buffer_size(this->get()); - } + class UHD_API phony_zero_copy_recv_if : public virtual zero_copy_if{ + public: - /*! - * Get a pointer to the underlying buffer. - * \return a pointer into memory - */ - template <class T> T cast(void) const{ - return boost::asio::buffer_cast<T>(this->get()); - } + //! structors + phony_zero_copy_recv_if(size_t max_buff_size); + ~phony_zero_copy_recv_if(void); -private: - /*! - * Get a reference to the internal mutable buffer. - * The buffer has a reference to memory and a size. - * \return a boost asio mutable buffer - */ - virtual const boost::asio::mutable_buffer &get(void) const = 0; -}; - -/*! - * A zero-copy interface for transport objects. - * Provides a way to get send and receive buffers - * with memory managed by the transport object. - */ -class UHD_API zero_copy_if : boost::noncopyable{ -public: - typedef boost::shared_ptr<zero_copy_if> sptr; + /*! + * Get a new receive buffer from this transport object. + */ + managed_recv_buffer::sptr get_recv_buff(void); - /*! - * Get a new receive buffer from this transport object. - */ - virtual managed_recv_buffer::sptr get_recv_buff(void) = 0; + private: + + /*! + * Perform a private copying recv. + * \param buff the buffer to write data into + * \return the number of bytes written to buff + */ + virtual size_t recv(const boost::asio::mutable_buffer &buff) = 0; + + struct impl; impl *_impl; //private implementation details + }; /*! - * Get a new send buffer from this transport object. + * A phony-zero-copy interface for transport objects that + * provides a zero-copy interface on top of copying transport. + * This interface implements the get managed send buffer, + * the base class must implement the private send method. */ - virtual managed_send_buffer::sptr get_send_buff(void) = 0; -}; + class UHD_API phony_zero_copy_send_if : public virtual zero_copy_if{ + public: + + //! structors + phony_zero_copy_send_if(size_t max_buff_size); + ~phony_zero_copy_send_if(void); + + /*! + * Get a new send buffer from this transport object. + */ + managed_send_buffer::sptr get_send_buff(void); + + private: + + /*! + * Perform a private copying send. + * \param buff the buffer to read data from + * \return the number of bytes read from buff + */ + virtual size_t send(const boost::asio::const_buffer &buff) = 0; + + struct impl; impl *_impl; //private implementation details + }; }} //namespace diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt index ed8c35225..a74f7d527 100644 --- a/host/lib/transport/CMakeLists.txt +++ b/host/lib/transport/CMakeLists.txt @@ -50,4 +50,5 @@ LIBUHD_APPEND_SOURCES( ${CMAKE_SOURCE_DIR}/lib/transport/udp_simple.cpp ${CMAKE_SOURCE_DIR}/lib/transport/udp_zero_copy_asio.cpp ${CMAKE_SOURCE_DIR}/lib/transport/vrt_packet_handler.hpp + ${CMAKE_SOURCE_DIR}/lib/transport/zero_copy.cpp ) diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index d6b3a8336..0c604811a 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -32,63 +32,17 @@ static const size_t MAX_DGRAM_SIZE = 2048; //assume max size on send and recv static const double RECV_TIMEOUT = 0.1; // 100 ms /*********************************************************************** - * Managed receive buffer implementation for udp zero-copy asio: - **********************************************************************/ -class managed_recv_buffer_impl : public managed_recv_buffer{ -public: - managed_recv_buffer_impl(const boost::asio::const_buffer &buff) : _buff(buff){ - /* NOP */ - } - - ~managed_recv_buffer_impl(void){ - delete [] this->cast<const boost::uint8_t *>(); - } - -private: - const boost::asio::const_buffer &get(void) const{ - return _buff; - } - - const boost::asio::const_buffer _buff; -}; - -/*********************************************************************** - * Managed send buffer implementation for udp zero-copy asio: - **********************************************************************/ -class managed_send_buffer_impl : public managed_send_buffer{ -public: - managed_send_buffer_impl( - const boost::asio::mutable_buffer &buff, - boost::asio::ip::udp::socket *socket - ) : _buff(buff), _socket(socket){ - /* NOP */ - } - - ~managed_send_buffer_impl(void){ - /* NOP */ - } - - void commit(size_t num_bytes){ - _socket->send(boost::asio::buffer(_buff, num_bytes)); - } - -private: - const boost::asio::mutable_buffer &get(void) const{ - return _buff; - } - - const boost::asio::mutable_buffer _buff; - boost::asio::ip::udp::socket *_socket; -}; - -/*********************************************************************** * Zero Copy UDP implementation with ASIO: * This is the portable zero copy implementation for systems * where a faster, platform specific solution is not available. * However, it is not a true zero copy implementation as each * send and recv requires a copy operation to/from userspace. **********************************************************************/ -class udp_zero_copy_impl : public udp_zero_copy{ +class udp_zero_copy_impl: + public virtual phony_zero_copy_recv_if, + public virtual phony_zero_copy_send_if, + public virtual udp_zero_copy +{ public: typedef boost::shared_ptr<udp_zero_copy_impl> sptr; @@ -96,17 +50,14 @@ public: udp_zero_copy_impl(const std::string &addr, const std::string &port); ~udp_zero_copy_impl(void); - //send/recv - managed_recv_buffer::sptr get_recv_buff(void); - managed_send_buffer::sptr get_send_buff(void); - - //manage buffer + //get size for internal socket buffer template <typename Opt> size_t get_buff_size(void){ Opt option; _socket->get_option(option); return option.value(); } + //set size for internal socket buffer template <typename Opt> size_t resize_buff(size_t num_bytes){ Opt option(num_bytes); _socket->set_option(option); @@ -117,13 +68,19 @@ private: boost::asio::ip::udp::socket *_socket; boost::asio::io_service _io_service; - //send and recv buffer memory (allocated once) - boost::uint8_t _send_mem[MIN_SOCK_BUFF_SIZE]; + size_t recv(const boost::asio::mutable_buffer &buff){ + return _socket->receive(boost::asio::buffer(buff)); + } - managed_send_buffer::sptr _send_buff; + size_t send(const boost::asio::const_buffer &buff){ + return _socket->send(boost::asio::buffer(buff)); + } }; -udp_zero_copy_impl::udp_zero_copy_impl(const std::string &addr, const std::string &port){ +udp_zero_copy_impl::udp_zero_copy_impl(const std::string &addr, const std::string &port): + phony_zero_copy_recv_if(MAX_DGRAM_SIZE), + phony_zero_copy_send_if(MAX_DGRAM_SIZE) +{ //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl; // resolve the address @@ -136,11 +93,6 @@ udp_zero_copy_impl::udp_zero_copy_impl(const std::string &addr, const std::strin _socket->open(boost::asio::ip::udp::v4()); _socket->connect(receiver_endpoint); - // create the managed send buff (just once) - _send_buff = managed_send_buffer::sptr(new managed_send_buffer_impl( - boost::asio::buffer(_send_mem, MIN_SOCK_BUFF_SIZE), _socket - )); - // set recv timeout timeval tv; tv.tv_sec = 0; @@ -156,23 +108,6 @@ udp_zero_copy_impl::~udp_zero_copy_impl(void){ delete _socket; } -managed_recv_buffer::sptr udp_zero_copy_impl::get_recv_buff(void){ - //allocate memory - boost::uint8_t *recv_mem = new boost::uint8_t[MAX_DGRAM_SIZE]; - - //call recv() with timeout option - size_t num_bytes = _socket->receive(boost::asio::buffer(recv_mem, MAX_DGRAM_SIZE)); - - //create a new managed buffer to house the data - return managed_recv_buffer::sptr( - new managed_recv_buffer_impl(boost::asio::buffer(recv_mem, num_bytes)) - ); -} - -managed_send_buffer::sptr udp_zero_copy_impl::get_send_buff(void){ - return _send_buff; //FIXME there is only ever one send buff, we assume that the caller doesnt hang onto these -} - /*********************************************************************** * UDP zero copy make function **********************************************************************/ diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp new file mode 100644 index 000000000..f69fd2774 --- /dev/null +++ b/host/lib/transport/zero_copy.cpp @@ -0,0 +1,140 @@ +// +// Copyright 2010 Ettus Research LLC +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see <http://www.gnu.org/licenses/>. +// + +#include <uhd/transport/zero_copy.hpp> +#include <boost/cstdint.hpp> +#include <boost/function.hpp> +#include <boost/bind.hpp> + +using namespace uhd::transport; + +/*********************************************************************** + * The pure-virtual deconstructor needs an implementation to be happy + **********************************************************************/ +managed_recv_buffer::~managed_recv_buffer(void){ + /* NOP */ +} + +/*********************************************************************** + * Phony zero-copy recv interface implementation + **********************************************************************/ + +//! phony zero-copy recv buffer implementation +class managed_recv_buffer_impl : public managed_recv_buffer{ +public: + managed_recv_buffer_impl(const boost::asio::const_buffer &buff) : _buff(buff){ + /* NOP */ + } + + ~managed_recv_buffer_impl(void){ + delete [] this->cast<const boost::uint8_t *>(); + } + +private: + const boost::asio::const_buffer &get(void) const{ + return _buff; + } + + const boost::asio::const_buffer _buff; +}; + +//! phony zero-copy recv interface implementation +struct phony_zero_copy_recv_if::impl{ + size_t max_buff_size; +}; + +phony_zero_copy_recv_if::phony_zero_copy_recv_if(size_t max_buff_size){ + _impl = new impl; + _impl->max_buff_size = max_buff_size; +} + +phony_zero_copy_recv_if::~phony_zero_copy_recv_if(void){ + delete _impl; +} + +managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(void){ + //allocate memory + boost::uint8_t *recv_mem = new boost::uint8_t[_impl->max_buff_size]; + + //call recv() with timeout option + size_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size)); + + //create a new managed buffer to house the data + return managed_recv_buffer::sptr( + new managed_recv_buffer_impl(boost::asio::buffer(recv_mem, num_bytes)) + ); +} + +/*********************************************************************** + * Phony zero-copy send interface implementation + **********************************************************************/ + +//! phony zero-copy send buffer implementation +class managed_send_buffer_impl : public managed_send_buffer{ +public: + typedef boost::function<size_t(const boost::asio::const_buffer &)> send_fcn_t; + + managed_send_buffer_impl( + const boost::asio::mutable_buffer &buff, + const send_fcn_t &send_fcn + ): + _buff(buff), + _send_fcn(send_fcn) + { + /* NOP */ + } + + ~managed_send_buffer_impl(void){ + /* NOP */ + } + + void commit(size_t num_bytes){ + _send_fcn(boost::asio::buffer(_buff, num_bytes)); + } + +private: + const boost::asio::mutable_buffer &get(void) const{ + return _buff; + } + + const boost::asio::mutable_buffer _buff; + const send_fcn_t _send_fcn; +}; + +//! phony zero-copy send interface implementation +struct phony_zero_copy_send_if::impl{ + boost::uint8_t *send_mem; + managed_send_buffer::sptr send_buff; +}; + +phony_zero_copy_send_if::phony_zero_copy_send_if(size_t max_buff_size){ + _impl = new impl; + _impl->send_mem = new boost::uint8_t[max_buff_size]; + _impl->send_buff = managed_send_buffer::sptr(new managed_send_buffer_impl( + boost::asio::buffer(_impl->send_mem, max_buff_size), + boost::bind(&phony_zero_copy_send_if::send, this, _1) + )); +} + +phony_zero_copy_send_if::~phony_zero_copy_send_if(void){ + delete [] _impl->send_mem; + delete _impl; +} + +managed_send_buffer::sptr phony_zero_copy_send_if::get_send_buff(void){ + return _impl->send_buff; //FIXME there is only ever one send buff, we assume that the caller doesnt hang onto these +} |