From f97c4338458a965502d73903983ae5a3ceeebd53 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Thu, 23 Sep 2010 18:07:31 -0700 Subject: usrp1: fixes to remove warnings and errors for usrp1 + libusb windows --- host/lib/transport/libusb1_zero_copy.cpp | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index b890a87f9..b3a160462 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -18,7 +18,6 @@ #include "libusb1_base.hpp" #include #include -#include #include #include #include @@ -208,7 +207,7 @@ libusb_transfer *usb_endpoint::allocate_transfer(int buff_len) endpoint, // endpoint buff, // buffer buff_len, // length - callback, // callback + libusb_transfer_cb_fn(callback), // callback this, // user_data 0); // timeout return lut; -- cgit v1.2.3 From dc8bcfde805228ed5d00334ce44c6c0684dcfe2c Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sat, 25 Sep 2010 21:07:15 -0700 Subject: usb: work on libusb code to use a single context across all calls libusb allocation stuff had been moved inside of smart pointer classes to handle automatic cleanup the public device handle implementation now holds an actual libusb device inside of it needs testing - all platforms --- host/include/uhd/transport/usb_device_handle.hpp | 6 - host/lib/transport/CMakeLists.txt | 1 - host/lib/transport/libusb1_base.cpp | 289 +++++++++++++++++------ host/lib/transport/libusb1_base.hpp | 162 ++++++++----- host/lib/transport/libusb1_control.cpp | 74 ++---- host/lib/transport/libusb1_device_handle.cpp | 118 --------- host/lib/transport/libusb1_zero_copy.cpp | 83 +++---- 7 files changed, 367 insertions(+), 366 deletions(-) delete mode 100644 host/lib/transport/libusb1_device_handle.cpp (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/include/uhd/transport/usb_device_handle.hpp b/host/include/uhd/transport/usb_device_handle.hpp index 735a3acbe..9bb7db9c4 100644 --- a/host/include/uhd/transport/usb_device_handle.hpp +++ b/host/include/uhd/transport/usb_device_handle.hpp @@ -60,12 +60,6 @@ public: */ virtual boost::uint16_t get_product_id() const = 0; - /*! - * Return the device's USB address - * \return a Product ID - */ - virtual boost::uint16_t get_device_addr() const = 0; - /*! * Return a vector of USB devices on this host * \return a vector of USB device handles that match vid and pid diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt index 48b6e9677..61616d077 100644 --- a/host/lib/transport/CMakeLists.txt +++ b/host/lib/transport/CMakeLists.txt @@ -31,7 +31,6 @@ IF(LIBUSB_FOUND) ${CMAKE_SOURCE_DIR}/lib/transport/libusb1_zero_copy.cpp ${CMAKE_SOURCE_DIR}/lib/transport/libusb1_base.cpp ${CMAKE_SOURCE_DIR}/lib/transport/libusb1_base.hpp - ${CMAKE_SOURCE_DIR}/lib/transport/libusb1_device_handle.cpp ) IF(MSVC) #include our custom stdint for libusb INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/lib/transport/msvc) diff --git a/host/lib/transport/libusb1_base.cpp b/host/lib/transport/libusb1_base.cpp index cd3e4adcf..5ff996642 100644 --- a/host/lib/transport/libusb1_base.cpp +++ b/host/lib/transport/libusb1_base.cpp @@ -17,107 +17,242 @@ #include "libusb1_base.hpp" #include +#include +#include +#include #include using namespace uhd::transport; -/********************************************************** - * Helper Methods - **********************************************************/ - -/********************************************************** - * libusb namespace - **********************************************************/ -void libusb::init(libusb_context **ctx, int debug_level) -{ - UHD_ASSERT_THROW(libusb_init(ctx) == 0); - libusb_set_debug(*ctx, debug_level); +/*********************************************************************** + * libusb session + **********************************************************************/ +class libusb_session_impl : public libusb::session{ +public: + libusb_session_impl(void){ + UHD_ASSERT_THROW(libusb_init(&_context) == 0); + libusb_set_debug(_context, debug_level); + } + + ~libusb_session_impl(void){ + libusb_exit(_context); + } + + libusb_context *get_context(void) const{ + return _context; + } + +private: + libusb_context *_context; +}; + +libusb::session::sptr libusb::session::get_global_session(void){ + static boost::weak_ptr global_session; + + //not expired -> get existing session + if (not global_session.expired()) return global_session.lock(); + + //create a new global session + sptr new_global_session(new libusb_session_impl()); + global_session = new_global_session; + return new_global_session; } -libusb_device_handle *libusb::open_device(libusb_context *ctx, - usb_device_handle::sptr handle) -{ - libusb_device_handle *dev_handle = NULL; - libusb_device **libusb_dev_list; - size_t dev_cnt = libusb_get_device_list(ctx, &libusb_dev_list); +/*********************************************************************** + * libusb device + **********************************************************************/ +class libusb_device_impl : public libusb::device{ +public: + libusb_device_impl(libusb_device *dev){ + _session = libusb::session::get_global_session(); + _dev = dev; + } - //find and open the USB device - for (size_t i = 0; i < dev_cnt; i++) { - libusb_device *dev = libusb_dev_list[i]; + ~libusb_device_impl(void){ + libusb_unref_device(this->get()); + } - if (compare_device(dev, handle)) { - if (libusb_open(dev, &dev_handle) == 0) break; - } + libusb_device *get(void) const{ + return _dev; } - libusb_free_device_list(libusb_dev_list, true); - if(dev_handle == NULL) - throw std::runtime_error("USB: cannot open device handle"); - return dev_handle; -} +private: + libusb::session::sptr _session; //always keep a reference to session + libusb_device *_dev; +}; + +/*********************************************************************** + * libusb device list + **********************************************************************/ +class libusb_device_list_impl : public libusb::device_list{ +public: + libusb_device_list_impl(void){ + libusb::session::sptr sess = libusb::session::get_global_session(); -//note: changed order of checks so it only tries to get_serial and get_device_address if vid and pid match -//doing this so it doesn't try to open the device if it's not ours -bool libusb::compare_device(libusb_device *dev, - usb_device_handle::sptr handle) -{ - std::string serial = handle->get_serial(); - boost::uint16_t vendor_id = handle->get_vendor_id(); - boost::uint16_t product_id = handle->get_product_id(); - boost::uint16_t device_addr = handle->get_device_addr(); - - libusb_device_descriptor libusb_desc; - if (libusb_get_device_descriptor(dev, &libusb_desc) != 0) - return false; - if (vendor_id != libusb_desc.idVendor) - return false; - if (product_id != libusb_desc.idProduct) - return false; - if (serial != get_serial(dev)) - return false; - if (device_addr != libusb_get_device_address(dev)) - return false; - - return true; + //allocate a new list of devices + libusb_device** dev_list; + ssize_t ret = libusb_get_device_list(sess->get_context(), &dev_list); + if (ret < 0) throw std::runtime_error("cannot enumerate usb devices"); + + //fill the vector of device references + for (size_t i = 0; i < size_t(ret); i++) _devs.push_back( + libusb::device::sptr(new libusb_device_impl(dev_list[i])) + ); + + //free the device list but dont unref (done in ~device) + libusb_free_device_list(dev_list, false/*dont unref*/); + } + + size_t size(void) const{ + return _devs.size(); + } + + libusb::device::sptr at(size_t i) const{ + return _devs.at(i); + } + +private: + std::vector _devs; +}; + +libusb::device_list::sptr libusb::device_list::make(void){ + return sptr(new libusb_device_list_impl()); } +/*********************************************************************** + * libusb device descriptor + **********************************************************************/ +class libusb_device_descriptor_impl : public libusb::device_descriptor{ +public: + libusb_device_descriptor_impl(libusb::device::sptr dev){ + _dev = dev; + UHD_ASSERT_THROW(libusb_get_device_descriptor(_dev->get(), &_desc) == 0); + } -bool libusb::open_interface(libusb_device_handle *dev_handle, - int interface) -{ - int ret = libusb_claim_interface(dev_handle, interface); - if (ret < 0) { - std::cerr << "error: libusb_claim_interface() " << ret << std::endl; - return false; + const libusb_device_descriptor &get(void) const{ + return _desc; } - else { - return true; + + std::string get_ascii_serial(void) const{ + if (this->get().iSerialNumber == 0) return ""; + + libusb::device_handle::sptr handle( + libusb::device_handle::get_cached_handle(_dev) + ); + + unsigned char buff[512]; + ssize_t ret = libusb_get_string_descriptor_ascii( + handle->get(), this->get().iSerialNumber, buff, sizeof(buff) + ); + if (ret < 0) return ""; //on error, just return empty string + + return std::string((char *)buff, ret); } + +private: + libusb::device::sptr _dev; //always keep a reference to device + libusb_device_descriptor _desc; +}; + +libusb::device_descriptor::sptr libusb::device_descriptor::make(device::sptr dev){ + return sptr(new libusb_device_descriptor_impl(dev)); } +/*********************************************************************** + * libusb device handle + **********************************************************************/ +class libusb_device_handle_impl : public libusb::device_handle{ +public: + libusb_device_handle_impl(libusb::device::sptr dev){ + _dev = dev; + UHD_ASSERT_THROW(libusb_open(_dev->get(), &_handle) == 0); + } + + ~libusb_device_handle_impl(void){ + //release all claimed interfaces + for (size_t i = 0; i < _claimed.size(); i++){ + libusb_release_interface(this->get(), _claimed[i]); + } + libusb_close(_handle); + } + + libusb_device_handle *get(void) const{ + return _handle; + } -std::string libusb::get_serial(libusb_device *dev) -{ - unsigned char buff[32]; + void claim_interface(int interface){ + UHD_ASSERT_THROW(libusb_claim_interface(this->get(), interface) == 0); + _claimed.push_back(interface); + } - libusb_device_descriptor desc; - if (libusb_get_device_descriptor(dev, &desc) < 0) - return ""; +private: + libusb::device::sptr _dev; //always keep a reference to device + libusb_device_handle *_handle; + std::vector _claimed; +}; - if (desc.iSerialNumber == 0) - return ""; +libusb::device_handle::sptr libusb::device_handle::get_cached_handle(device::sptr dev){ + static uhd::dict > handles; - //open the device because we have to - libusb_device_handle *dev_handle; - if (libusb_open(dev, &dev_handle) < 0) - return ""; + //not expired -> get existing session + if (handles.has_key(dev->get()) and not handles[dev->get()].expired()){ + return handles[dev->get()].lock(); + } - if (libusb_get_string_descriptor_ascii(dev_handle, desc.iSerialNumber, - buff, sizeof(buff)) < 0) { - return ""; + //create a new global session + sptr new_handle(new libusb_device_handle_impl(dev)); + handles[dev->get()] = new_handle; + return new_handle; +} + +/*********************************************************************** + * libusb special handle + **********************************************************************/ +class libusb_special_handle_impl : public libusb::special_handle{ +public: + libusb_special_handle_impl(libusb::device::sptr dev){ + _dev = dev; } - libusb_close(dev_handle); + libusb::device::sptr get_device(void) const{ + return _dev; + } + + std::string get_serial(void) const{ + return libusb::device_descriptor::make(this->get_device())->get_ascii_serial(); + } + + boost::uint16_t get_vendor_id(void) const{ + return libusb::device_descriptor::make(this->get_device())->get().idVendor; + } + + boost::uint16_t get_product_id(void) const{ + return libusb::device_descriptor::make(this->get_device())->get().idProduct; + } + +private: + libusb::device::sptr _dev; //always keep a reference to device +}; + +libusb::special_handle::sptr libusb::special_handle::make(device::sptr dev){ + return sptr(new libusb_special_handle_impl(dev)); +} + +/*********************************************************************** + * list device handles implementations + **********************************************************************/ +std::vector usb_device_handle::get_device_list( + boost::uint16_t vid, boost::uint16_t pid +){ + std::vector handles; + + libusb::device_list::sptr dev_list = libusb::device_list::make(); + for (size_t i = 0; i < dev_list->size(); i++){ + usb_device_handle::sptr handle = libusb::special_handle::make(dev_list->at(i)); + if (handle->get_vendor_id() == vid and handle->get_product_id() == pid){ + handles.push_back(handle); + } + } - return (char*) buff; + return handles; } diff --git a/host/lib/transport/libusb1_base.hpp b/host/lib/transport/libusb1_base.hpp index 484bcf3d9..47d078cde 100644 --- a/host/lib/transport/libusb1_base.hpp +++ b/host/lib/transport/libusb1_base.hpp @@ -19,74 +19,122 @@ #define INCLUDED_LIBUHD_TRANSPORT_LIBUSB_HPP #include +#include +#include #include #include namespace uhd { namespace transport { namespace libusb { - /* - * Initialize libusb and set debug level - * Takes a pointer to context pointer because that's - * how libusb rolls. Debug levels. - * - * Level 0: no messages ever printed by the library (default) - * Level 1: error messages are printed to stderr - * Level 2: warning and error messages are printed to stderr - * Level 3: informational messages are printed to stdout, warning - * and error messages are printed to stderr - * - * \param ctx pointer to context pointer - * \param debug_level + + /*! + * This session class holds a global libusb context for this process. + * The get global session call will create a new context if none exists. + * When all references to session are destroyed, the context will be freed. */ - void init(libusb_context **ctx, int debug_level); - - /* - * Open the device specified by a generic handle - * Find the libusb_device cooresponding to the generic handle - * and open it for I/O, which returns a libusb_device_handle - * ready for an interface - * \param ctx the libusb context used for init - * \return a libusb_device_handle ready for action + class session : boost::noncopyable { + public: + typedef boost::shared_ptr sptr; + + /*! + * Level 0: no messages ever printed by the library (default) + * Level 1: error messages are printed to stderr + * Level 2: warning and error messages are printed to stderr + * Level 3: informational messages are printed to stdout, warning + * and error messages are printed to stderr + */ + static const int debug_level = 0; + + //! get a shared pointer to the global session + static sptr get_global_session(void); + + //! get the underlying libusb context pointer + virtual libusb_context *get_context(void) const = 0; + }; + + /*! + * Holds a device pointer with a reference to the session. */ - libusb_device_handle *open_device(libusb_context *ctx, - usb_device_handle::sptr handle); - - /* - * Compare a libusb device with a generic handle - * Check the descriptors and open the device to check the - * serial number string. Compare values against the given - * handle. The libusb context is already implied in the - * libusb_device. - * \param dev a libusb_device pointer - * \param handle a generic handle specifier - * \return true if handle and device match, false otherwise + class device : boost::noncopyable { + public: + typedef boost::shared_ptr sptr; + + //! get the underlying device pointer + virtual libusb_device *get(void) const = 0; + }; + + /*! + * This device list class holds a device list that will be + * automatically freed when the last reference is destroyed. */ - bool compare_device(libusb_device *dev, usb_device_handle::sptr handle); - - /* - * Open an interface to the device - * This is a logical operation for operating system housekeeping as - * nothing is sent over the bus. The interface much correspond - * to the USB device descriptors. - * \param dev_handle libusb handle to an opened device - * \param interface integer of the interface to use - * \return true on success, false on error + class device_list : boost::noncopyable { + public: + typedef boost::shared_ptr sptr; + + //! make a new device list + static sptr make(void); + + //! the number of devices in this list + virtual size_t size() const = 0; + + //! get the device pointer at a particular index + virtual device::sptr at(size_t index) const = 0; + }; + + /*! + * Holds a device descriptor and a reference to the device. */ - bool open_interface(libusb_device_handle *dev_handle, int interface); - - /* - * Get serial number - * The standard USB device descriptor contains an index to an - * actual serial number string descriptor. The index is readily - * readble, but the string descriptor requires probing the device. - * Because this call attempts to open the device, it may not - * succeed because not all USB devices are readily opened. - * The default language is used for the request (English). - * \param dev a libusb_device pointer - * \return string serial number or 0 on error or unavailablity + class device_descriptor : boost::noncopyable { + public: + typedef boost::shared_ptr sptr; + + //! make a new descriptor from a device reference + static sptr make(device::sptr); + + //! get the underlying device descriptor + virtual const libusb_device_descriptor &get(void) const = 0; + + virtual std::string get_ascii_serial(void) const = 0; + }; + + /*! + * Holds a device handle and a reference to the device. */ - std::string get_serial(libusb_device *dev); + class device_handle : boost::noncopyable { + public: + typedef boost::shared_ptr sptr; + + //! get a cached handle or make a new one given the device + static sptr get_cached_handle(device::sptr); + + //! get the underlying device handle + virtual libusb_device_handle *get(void) const = 0; + + /*! + * Open USB interfaces for control using magic value + * IN interface: 2 + * OUT interface: 1 + * Control interface: 0 + */ + virtual void claim_interface(int) = 0; + }; + + /*! + * The special handle is our internal implementation of the + * usb device handle which is used publicly to identify a device. + */ + class special_handle : public usb_device_handle { + public: + typedef boost::shared_ptr sptr; + + //! make a new special handle from device + static sptr make(device::sptr); + + //! get the underlying device reference + virtual device::sptr get_device(void) const = 0; + }; + } }} //namespace diff --git a/host/lib/transport/libusb1_control.cpp b/host/lib/transport/libusb1_control.cpp index be9b51111..c989a788c 100644 --- a/host/lib/transport/libusb1_control.cpp +++ b/host/lib/transport/libusb1_control.cpp @@ -20,7 +20,6 @@ using namespace uhd::transport; -const int libusb_debug_level = 0; const int libusb_timeout = 0; /*********************************************************************** @@ -28,69 +27,38 @@ const int libusb_timeout = 0; **********************************************************************/ class libusb_control_impl : public usb_control { public: - libusb_control_impl(usb_device_handle::sptr handle); - ~libusb_control_impl(); + libusb_control_impl(libusb::device_handle::sptr handle): + _handle(handle) + { + _handle->claim_interface(0 /* control interface */); + } size_t submit(boost::uint8_t request_type, boost::uint8_t request, boost::uint16_t value, boost::uint16_t index, unsigned char *buff, - boost::uint16_t length); + boost::uint16_t length + ){ + return libusb_control_transfer(_handle->get(), + request_type, + request, + value, + index, + buff, + length, + libusb_timeout); + } private: - libusb_context *_ctx; - libusb_device_handle *_dev_handle; + libusb::device_handle::sptr _handle; }; - -libusb_control_impl::libusb_control_impl(usb_device_handle::sptr handle) -{ - _ctx = NULL; - libusb::init(&_ctx, libusb_debug_level); - - // Find and open the libusb_device corresponding to the - // given handle and return the libusb_device_handle - // that can be used for I/O purposes. - _dev_handle = libusb::open_device(_ctx, handle); - - // Open USB interfaces for control using magic value - // IN interface: 2 - // OUT interface: 1 - // Control interface: 0 - libusb::open_interface(_dev_handle, 0); -} - - -libusb_control_impl::~libusb_control_impl() -{ - libusb_close(_dev_handle); - libusb_exit(_ctx); -} - - -size_t libusb_control_impl::submit(boost::uint8_t request_type, - boost::uint8_t request, - boost::uint16_t value, - boost::uint16_t index, - unsigned char *buff, - boost::uint16_t length) -{ - return libusb_control_transfer(_dev_handle, - request_type, - request, - value, - index, - buff, - length, - libusb_timeout); -} - - /*********************************************************************** * USB control public make functions **********************************************************************/ -usb_control::sptr usb_control::make(usb_device_handle::sptr handle) -{ - return sptr(new libusb_control_impl(handle)); +usb_control::sptr usb_control::make(usb_device_handle::sptr handle){ + return sptr(new libusb_control_impl(libusb::device_handle::get_cached_handle( + boost::static_pointer_cast(handle)->get_device() + ))); } diff --git a/host/lib/transport/libusb1_device_handle.cpp b/host/lib/transport/libusb1_device_handle.cpp deleted file mode 100644 index 6bef37ed2..000000000 --- a/host/lib/transport/libusb1_device_handle.cpp +++ /dev/null @@ -1,118 +0,0 @@ -// -// 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 . -// - -#include "libusb1_base.hpp" -#include -#include - -using namespace uhd::transport; - -const int libusb_debug_level = 0; - -/**************************************************************** - * libusb USB device handle implementation class - ***************************************************************/ -class libusb1_device_handle_impl : public usb_device_handle { -public: - libusb1_device_handle_impl(std::string serial, - boost::uint16_t product_id, - boost::uint16_t vendor_id, - boost::uint16_t device_addr) - : _serial(serial), _product_id(product_id), - _vendor_id(vendor_id), _device_addr(device_addr) - { - /* NOP */ - } - - ~libusb1_device_handle_impl() - { - /* NOP */ - } - - std::string get_serial() const - { - return _serial; - } - - boost::uint16_t get_vendor_id() const - { - return _vendor_id; - } - - - boost::uint16_t get_product_id() const - { - return _product_id; - } - - boost::uint16_t get_device_addr() const - { - return _device_addr; - } - -private: - std::string _serial; - boost::uint16_t _product_id; - boost::uint16_t _vendor_id; - boost::uint16_t _device_addr; -}; - - -usb_device_handle::sptr make_usb_device_handle(libusb_device *dev) -{ - libusb_device_descriptor desc; - - if (libusb_get_device_descriptor(dev, &desc) < 0) { - throw std::runtime_error("USB: failed to get device descriptor"); - } - - std::string serial = libusb::get_serial(dev); - boost::uint16_t product_id = desc.idProduct; - boost::uint16_t vendor_id = desc.idVendor; - boost::uint16_t device_addr = libusb_get_device_address(dev); - - return usb_device_handle::sptr(new libusb1_device_handle_impl( - serial, - product_id, - vendor_id, - device_addr)); -} - -std::vector usb_device_handle::get_device_list(boost::uint16_t vid, boost::uint16_t pid) -{ - libusb_context *ctx = NULL; - libusb_device** libusb_device_list; - std::vector device_handle_list; - libusb_device_descriptor desc; - - libusb::init(&ctx, libusb_debug_level); - - size_t dev_size = libusb_get_device_list(ctx, &libusb_device_list); - for (size_t i = 0; i < dev_size; i++) { - libusb_device *dev = libusb_device_list[i]; - if(libusb_get_device_descriptor(dev, &desc) != 0) { - continue; //just try the next device, do not throw - } - if(desc.idVendor == vid && desc.idProduct == pid) { - device_handle_list.push_back(make_usb_device_handle(dev)); - } - } - - libusb_free_device_list(libusb_device_list, true); - libusb_exit(ctx); - return device_handle_list; -} diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index b3a160462..41cd4b3fc 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -24,7 +24,6 @@ using namespace uhd::transport; -const int libusb_debug_level = 0; const int libusb_timeout = 0; /*********************************************************************** @@ -57,8 +56,8 @@ void pp_transfer(libusb_transfer *lut) **********************************************************************/ class usb_endpoint { private: - libusb_device_handle *_dev_handle; - libusb_context *_ctx; + libusb::device_handle::sptr _handle; + libusb::session::sptr _session; int _endpoint; bool _input; @@ -90,9 +89,8 @@ private: void print_transfer_status(libusb_transfer *lut); public: - usb_endpoint(libusb_device_handle *dev_handle, - libusb_context *ctx, int endpoint, bool input, - size_t transfer_size, size_t num_transfers); + usb_endpoint(libusb::device_handle::sptr handle, libusb::session::sptr sess, + int endpoint, bool input, size_t transfer_size, size_t num_transfers); ~usb_endpoint(); @@ -143,11 +141,11 @@ void usb_endpoint::callback_handle_transfer(libusb_transfer *lut) * submit the transfers so that they're ready to return when * data is available. */ -usb_endpoint::usb_endpoint(libusb_device_handle *dev_handle, - libusb_context *ctx, int endpoint, bool input, - size_t transfer_size, size_t num_transfers) - : _dev_handle(dev_handle), - _ctx(ctx), _endpoint(endpoint), _input(input), +usb_endpoint::usb_endpoint( + libusb::device_handle::sptr handle, libusb::session::sptr sess, + int endpoint, bool input, size_t transfer_size, size_t num_transfers) + : _handle(handle), _session(sess), + _endpoint(endpoint), _input(input), _transfer_size(transfer_size), _num_transfers(num_transfers) { unsigned int i; @@ -203,7 +201,7 @@ libusb_transfer *usb_endpoint::allocate_transfer(int buff_len) unsigned int endpoint = ((_endpoint & 0x7f) | (_input ? 0x80 : 0)); libusb_fill_bulk_transfer(lut, // transfer - _dev_handle, // dev_handle + _handle->get(), // dev_handle endpoint, // endpoint buff, // buffer buff_len, // length @@ -356,7 +354,7 @@ bool usb_endpoint::reap_pending_list() { int retval; - if ((retval = libusb_handle_events(_ctx)) < 0) { + if ((retval = libusb_handle_events(_session->get_context())) < 0) { std::cerr << "error: libusb_handle_events: " << retval << std::endl; return false; } @@ -383,7 +381,7 @@ bool usb_endpoint::reap_pending_list_timeout() size_t pending_list_size = _pending_list.size(); - if ((retval = libusb_handle_events_timeout(_ctx, &tv)) < 0) { + if ((retval = libusb_handle_events_timeout(_session->get_context(), &tv)) < 0) { std::cerr << "error: libusb_handle_events: " << retval << std::endl; return false; } @@ -613,11 +611,8 @@ private: usb_endpoint *_rx_ep; usb_endpoint *_tx_ep; - // Maintain libusb values - libusb_context *_rx_ctx; - libusb_context *_tx_ctx; - libusb_device_handle *_rx_dev_handle; - libusb_device_handle *_tx_dev_handle; + libusb::device_handle::sptr _handle; + libusb::session::sptr _session; size_t _recv_buff_size; size_t _send_buff_size; @@ -626,7 +621,7 @@ private: public: typedef boost::shared_ptr sptr; - libusb_zero_copy_impl(usb_device_handle::sptr handle, + libusb_zero_copy_impl(libusb::device_handle::sptr handle, unsigned int rx_endpoint, unsigned int tx_endpoint, size_t recv_buff_size, @@ -646,44 +641,27 @@ public: * Initializes libusb, opens devices, and sets up interfaces for I/O. * Finally, creates endpoints for asynchronous I/O. */ -libusb_zero_copy_impl::libusb_zero_copy_impl(usb_device_handle::sptr handle, +libusb_zero_copy_impl::libusb_zero_copy_impl(libusb::device_handle::sptr handle, unsigned int rx_endpoint, unsigned int tx_endpoint, size_t buff_size, size_t block_size) - : _rx_ctx(NULL), _tx_ctx(NULL), _rx_dev_handle(NULL), _tx_dev_handle(NULL), + : _handle(handle), _session(libusb::session::get_global_session()), _recv_buff_size(block_size), _send_buff_size(block_size), _num_frames(buff_size / block_size) { - // Initialize libusb with separate contexts to allow - // thread safe operation of transmit and receive - libusb::init(&_rx_ctx, libusb_debug_level); - libusb::init(&_tx_ctx, libusb_debug_level); - - UHD_ASSERT_THROW((_rx_ctx != NULL) && (_tx_ctx != NULL)); - - // Find and open the libusb_device corresponding to the - // given handle and return the libusb_device_handle - // that can be used for I/O purposes. - _rx_dev_handle = libusb::open_device(_rx_ctx, handle); - _tx_dev_handle = libusb::open_device(_tx_ctx, handle); - - // Open USB interfaces for tx/rx using magic values. - // IN interface: 2 - // OUT interface: 1 - // Control interface: 0 - libusb::open_interface(_rx_dev_handle, 2); - libusb::open_interface(_tx_dev_handle, 1); - - _rx_ep = new usb_endpoint(_rx_dev_handle, // libusb device_handle - _rx_ctx, // libusb context + _handle->claim_interface(2 /*in interface*/); + _handle->claim_interface(1 /*out interface*/); + + _rx_ep = new usb_endpoint(_handle, // libusb device_handle + _session, // libusb session w/ context rx_endpoint, // USB endpoint number true, // IN endpoint _recv_buff_size, // buffer size per transfer _num_frames); // number of libusb transfers - _tx_ep = new usb_endpoint(_tx_dev_handle, // libusb device_handle - _tx_ctx, // libusb context + _tx_ep = new usb_endpoint(_handle, // libusb device_handle + _session, // libusb session w/ context tx_endpoint, // USB endpoint number false, // OUT endpoint _send_buff_size, // buffer size per transfer @@ -694,13 +672,7 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(usb_device_handle::sptr handle, libusb_zero_copy_impl::~libusb_zero_copy_impl() { delete _rx_ep; - delete _tx_ep; - - libusb_close(_rx_dev_handle); - libusb_close(_tx_dev_handle); - - libusb_exit(_rx_ctx); - libusb_exit(_tx_ctx); + delete _tx_ep; } @@ -755,7 +727,10 @@ usb_zero_copy::sptr usb_zero_copy::make(usb_device_handle::sptr handle, size_t block_size) { - return sptr(new libusb_zero_copy_impl(handle, + libusb::device_handle::sptr dev_handle(libusb::device_handle::get_cached_handle( + boost::static_pointer_cast(handle)->get_device() + )); + return sptr(new libusb_zero_copy_impl(dev_handle, rx_endpoint, tx_endpoint, buff_size, -- cgit v1.2.3 From 0816925695aa69a1ae344863fef47d239b3ec8af Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sun, 26 Sep 2010 20:23:23 -0700 Subject: usb: zero copy work, multiple endpoints with single context, async io Heavy work on the zero copy interface and endpoint wrappers to properly use the async io. The global libusb session starts a thread to run the event handler, the async callbacks push completed transfers onto a thread-safe bounded buffer. The managed buffer creation routines use the bounded buffer to efficiently pop off completed transfers. works on linux, throws a weird exception on cleanup --- host/lib/transport/libusb1_base.cpp | 16 ++ host/lib/transport/libusb1_zero_copy.cpp | 474 ++++++++----------------------- 2 files changed, 134 insertions(+), 356 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/lib/transport/libusb1_base.cpp b/host/lib/transport/libusb1_base.cpp index 5ff996642..26e864459 100644 --- a/host/lib/transport/libusb1_base.cpp +++ b/host/lib/transport/libusb1_base.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include using namespace uhd::transport; @@ -32,9 +33,12 @@ public: libusb_session_impl(void){ UHD_ASSERT_THROW(libusb_init(&_context) == 0); libusb_set_debug(_context, debug_level); + _thread_group.create_thread(boost::bind(&libusb_session_impl::run_event_loop, this)); } ~libusb_session_impl(void){ + _running = false; + _thread_group.join_all(); libusb_exit(_context); } @@ -44,6 +48,18 @@ public: private: libusb_context *_context; + boost::thread_group _thread_group; + bool _running; + + void run_event_loop(void){ + _running = true; + timeval tv; + while(_running){ + tv.tv_sec = 0; + tv.tv_usec = 100000; //100ms + libusb_handle_events_timeout(this->get_context(), &tv); + } + } }; libusb::session::sptr libusb::session::get_global_session(void){ diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index 41cd4b3fc..e84793e88 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -17,8 +17,11 @@ #include "libusb1_base.hpp" #include +#include #include -#include +#include +#include +#include #include #include @@ -55,52 +58,57 @@ void pp_transfer(libusb_transfer *lut) * interface provided by the kernel. **********************************************************************/ class usb_endpoint { +public: + typedef boost::shared_ptr sptr; + + usb_endpoint( + libusb::device_handle::sptr handle, + int endpoint, + bool input, + size_t transfer_size, + size_t num_transfers + ); + + ~usb_endpoint(void); + + // Exposed interface for submitting / retrieving transfer buffers + + //! Submit a new transfer that was presumably just filled or emptied. + void submit(libusb_transfer *lut); + + /*! + * Get an available transfer: + * For inputs, this is a just filled transfer. + * For outputs, this is a just emptied transfer. + * \param timeout_ms the timeout to wait for a lut + * \return the transfer pointer or NULL if timeout + */ + libusb_transfer *get_lut_with_wait(size_t timeout_ms = 100); + + //Callback use only + void callback_handle_transfer(libusb_transfer *lut); + private: libusb::device_handle::sptr _handle; - libusb::session::sptr _session; int _endpoint; bool _input; size_t _transfer_size; size_t _num_transfers; - // Transfer state lists (transfers are free, pending, or completed) - std::list _free_list; - std::list _pending_list; - std::list _completed_list; + //! hold a bounded buffer of completed transfers + typedef bounded_buffer lut_buff_type; + lut_buff_type::sptr _completed_list; + + //! a list of all transfer structs we allocated + std::vector _all_luts; + + //! a list of shared arrays for the transfer buffers + std::vector > _buffers; // Calls for processing asynchronous I/O libusb_transfer *allocate_transfer(int buff_len); - bool cancel(libusb_transfer *lut); - bool cancel_all(); - bool reap_pending_list(); - bool reap_pending_list_timeout(); - bool reap_completed_list(); - - // Transfer state manipulators - void free_list_add(libusb_transfer *lut); - void pending_list_add(libusb_transfer *lut); - void completed_list_add(libusb_transfer *lut); - libusb_transfer *free_list_get(); - libusb_transfer *completed_list_get(); - bool pending_list_remove(libusb_transfer *lut); - - // Debug use void print_transfer_status(libusb_transfer *lut); - -public: - usb_endpoint(libusb::device_handle::sptr handle, libusb::session::sptr sess, - int endpoint, bool input, size_t transfer_size, size_t num_transfers); - - ~usb_endpoint(); - - // Exposed interface for submitting / retrieving transfer buffers - bool submit(libusb_transfer *lut); - libusb_transfer *get_completed_transfer(); - libusb_transfer *get_free_transfer(); - - //Callback use only - void callback_handle_transfer(libusb_transfer *lut); }; @@ -113,9 +121,8 @@ public: * it from the pending to completed status list. * \param lut pointer to libusb_transfer */ -static void callback(libusb_transfer *lut) -{ - usb_endpoint *endpoint = (usb_endpoint *) lut->user_data; +static void callback(libusb_transfer *lut){ + usb_endpoint *endpoint = (usb_endpoint *) lut->user_data; endpoint->callback_handle_transfer(lut); } @@ -124,14 +131,8 @@ static void callback(libusb_transfer *lut) * Accessor call to allow list access from callback space * \param pointer to libusb_transfer */ -void usb_endpoint::callback_handle_transfer(libusb_transfer *lut) -{ - if (!pending_list_remove(lut)) { - std::cerr << "USB: pending remove failed" << std::endl; - return; - } - - completed_list_add(lut); +void usb_endpoint::callback_handle_transfer(libusb_transfer *lut){ + _completed_list->push_with_wait(lut); } @@ -142,18 +143,27 @@ void usb_endpoint::callback_handle_transfer(libusb_transfer *lut) * data is available. */ usb_endpoint::usb_endpoint( - libusb::device_handle::sptr handle, libusb::session::sptr sess, - int endpoint, bool input, size_t transfer_size, size_t num_transfers) - : _handle(handle), _session(sess), - _endpoint(endpoint), _input(input), - _transfer_size(transfer_size), _num_transfers(num_transfers) + libusb::device_handle::sptr handle, + int endpoint, + bool input, + size_t transfer_size, + size_t num_transfers +): + _handle(handle), + _endpoint(endpoint), + _input(input), + _transfer_size(transfer_size), + _num_transfers(num_transfers) { - unsigned int i; - for (i = 0; i < _num_transfers; i++) { - free_list_add(allocate_transfer(_transfer_size)); + _completed_list = lut_buff_type::make(num_transfers); - if (_input) - submit(free_list_get()); + for (size_t i = 0; i < _num_transfers; i++){ + _all_luts.push_back(allocate_transfer(_transfer_size)); + + //input luts are immediately submitted to be filled + //output luts go into the completed list as free buffers + if (_input) this->submit(_all_luts.back()); + else _completed_list->push_with_wait(_all_luts.back()); } } @@ -165,22 +175,21 @@ usb_endpoint::usb_endpoint( * the transfers. Libusb will deallocate the data buffer held by * each transfer. */ -usb_endpoint::~usb_endpoint() -{ - cancel_all(); - - while (!_pending_list.empty()) { - if (!reap_pending_list()) - std::cerr << "error: destructor failed to reap" << std::endl; +usb_endpoint::~usb_endpoint(void){ + //cancel all transfers + BOOST_FOREACH(libusb_transfer *lut, _all_luts){ + libusb_cancel_transfer(lut); } - while (!_completed_list.empty()) { - if (!reap_completed_list()) - std::cerr << "error: destructor failed to reap" << std::endl; - } + //collect canceled transfers (drain the queue) + libusb_transfer *lut; + while(_completed_list->pop_with_timed_wait( + lut, boost::posix_time::milliseconds(100) + )); - while (!_free_list.empty()) { - libusb_free_transfer(free_list_get()); + //free all transfers + BOOST_FOREACH(libusb_transfer *lut, _all_luts){ + libusb_free_transfer(lut); } } @@ -192,20 +201,21 @@ usb_endpoint::~usb_endpoint() * \param buff_len size of the individual buffer held by each transfer * \return pointer to an allocated libusb_transfer */ -libusb_transfer *usb_endpoint::allocate_transfer(int buff_len) -{ +libusb_transfer *usb_endpoint::allocate_transfer(int buff_len){ libusb_transfer *lut = libusb_alloc_transfer(0); - unsigned char *buff = new unsigned char[buff_len]; + boost::shared_array buff(new boost::uint8_t[buff_len]); + _buffers.push_back(buff); //store a reference to this shared array unsigned int endpoint = ((_endpoint & 0x7f) | (_input ? 0x80 : 0)); + libusb_transfer_cb_fn lut_callback = libusb_transfer_cb_fn(&callback); libusb_fill_bulk_transfer(lut, // transfer _handle->get(), // dev_handle endpoint, // endpoint - buff, // buffer + buff.get(), // buffer buff_len, // length - libusb_transfer_cb_fn(callback), // callback + lut_callback, // callback this, // user_data 0); // timeout return lut; @@ -218,95 +228,15 @@ libusb_transfer *usb_endpoint::allocate_transfer(int buff_len) * \param lut pointer to libusb_transfer * \return true on success or false on error */ -bool usb_endpoint::submit(libusb_transfer *lut) -{ - int retval; - if ((retval = libusb_submit_transfer(lut)) < 0) { - std::cerr << "error: libusb_submit_transfer: " << retval << std::endl; - return false; - } - - pending_list_add(lut); - return true; -} - - -/* - * Cancel a pending transfer - * Search the pending list for the transfer and cancel if found. - * \param lut pointer to libusb_transfer to cancel - * \return true on success or false if transfer is not found - * - * Note: success only indicates submission of cancelation request. - * Sucessful cancelation is not known until the callback occurs. - */ -bool usb_endpoint::cancel(libusb_transfer *lut) -{ - std::list::iterator iter; - for (iter = _pending_list.begin(); iter != _pending_list.end(); iter++) { - if (*iter == lut) { - libusb_cancel_transfer(lut); - return true; - } - } - return false; -} - - -/* - * Cancel all pending transfers - * \return bool true if cancelation request is submitted - * - * Note: success only indicates submission of cancelation request. - * Sucessful cancelation is not known until the callback occurs. - */ -bool usb_endpoint::cancel_all() -{ - std::list::iterator iter; - - for (iter = _pending_list.begin(); iter != _pending_list.end(); iter++) { - if (libusb_cancel_transfer(*iter) < 0) { - std::cerr << "error: libusb_cancal_transfer() failed" << std::endl; - return false; - } - } - - return true; -} - - -/* - * Reap completed transfers - * return true if at least one transfer was reaped, false otherwise. - * Check completed transfers for errors and mark as free. This is a - * blocking call. - * \return bool true if a libusb transfer is reaped, false otherwise - */ -bool usb_endpoint::reap_completed_list() -{ - libusb_transfer *lut; - - if (_completed_list.empty()) { - if (!reap_pending_list_timeout()) - return false; - } - - while (!_completed_list.empty()) { - lut = completed_list_get(); - print_transfer_status(lut); - free_list_add(lut); - } - - return true; +void usb_endpoint::submit(libusb_transfer *lut){ + UHD_ASSERT_THROW(libusb_submit_transfer(lut) == 0); } - /* * Print status errors of a completed transfer * \param lut pointer to an libusb_transfer */ -void usb_endpoint::print_transfer_status(libusb_transfer *lut) -{ +void usb_endpoint::print_transfer_status(libusb_transfer *lut){ switch (lut->status) { case LIBUSB_TRANSFER_COMPLETED: if (lut->actual_length < lut->length) { @@ -342,166 +272,14 @@ void usb_endpoint::print_transfer_status(libusb_transfer *lut) } } - -/* - * Reap pending transfers without timeout - * This is a blocking call. Reaping submitted transfers is - * handled by libusb and the assigned callback function. - * Block until at least one transfer is reaped. - * \return true true if a transfer was reaped or false otherwise - */ -bool usb_endpoint::reap_pending_list() -{ - int retval; - - if ((retval = libusb_handle_events(_session->get_context())) < 0) { - std::cerr << "error: libusb_handle_events: " << retval << std::endl; - return false; - } - - return true; -} - - -/* - * Reap pending transfers with timeout - * This call blocks until a transfer is reaped or timeout. - * Reaping submitted transfers is handled by libusb and the - * assigned callback function. Block until at least one - * transfer is reaped or timeout occurs. - * \return true if a transfer was reaped or false otherwise - */ -bool usb_endpoint::reap_pending_list_timeout() -{ - int retval; - timeval tv; - - tv.tv_sec = 0; - tv.tv_usec = 100000; //100ms - - size_t pending_list_size = _pending_list.size(); - - if ((retval = libusb_handle_events_timeout(_session->get_context(), &tv)) < 0) { - std::cerr << "error: libusb_handle_events: " << retval << std::endl; - return false; - } - - if (_pending_list.size() < pending_list_size) { - return true; - } - else { - return false; - } -} - - -/* - * Get a free transfer - * The transfer has an empty data bufer for OUT requests - * \return pointer to a libusb_transfer - */ -libusb_transfer *usb_endpoint::get_free_transfer() -{ - if (_free_list.empty()) { - if (!reap_completed_list()) - return NULL; - } - - return free_list_get(); -} - - -/* - * Get a completed transfer - * The transfer has a full data buffer for IN requests - * \return pointer to libusb_transfer - */ -libusb_transfer *usb_endpoint::get_completed_transfer() -{ - if (_completed_list.empty()) { - if (!reap_pending_list_timeout()) - return NULL; - } - - return completed_list_get(); -} - -/* - * List operations - */ -void usb_endpoint::free_list_add(libusb_transfer *lut) -{ - _free_list.push_back(lut); -} - -void usb_endpoint::pending_list_add(libusb_transfer *lut) -{ - _pending_list.push_back(lut); -} - -void usb_endpoint::completed_list_add(libusb_transfer *lut) -{ - _completed_list.push_back(lut); -} - - -/* - * Free and completed lists don't have ordered content - * Pop transfers from the front as needed - */ -libusb_transfer *usb_endpoint::free_list_get() -{ - libusb_transfer *lut; - - if (_free_list.size() == 0) { - return NULL; - } - else { - lut = _free_list.front(); - _free_list.pop_front(); - return lut; - } -} - - -/* - * Free and completed lists don't have ordered content - * Pop transfers from the front as needed - */ -libusb_transfer *usb_endpoint::completed_list_get() -{ +libusb_transfer *usb_endpoint::get_lut_with_wait(size_t timeout_ms){ libusb_transfer *lut; - - if (_completed_list.empty()) { - return NULL; - } - else { - lut = _completed_list.front(); - _completed_list.pop_front(); - return lut; - } + if (_completed_list->pop_with_timed_wait( + lut, boost::posix_time::milliseconds(timeout_ms) + )) return lut; + return NULL; } - -/* - * Search and remove transfer from pending list - * Assuming that the callbacks occur in order, the front element - * should yield the correct transfer. If not, then something else - * is going on. If no transfers match, then something went wrong. - */ -bool usb_endpoint::pending_list_remove(libusb_transfer *lut) -{ - std::list::iterator iter; - for (iter = _pending_list.begin(); iter != _pending_list.end(); iter++) { - if (*iter == lut) { - _pending_list.erase(iter); - return true; - } - } - return false; -} - - /*********************************************************************** * Managed buffers **********************************************************************/ @@ -515,17 +293,15 @@ bool usb_endpoint::pending_list_remove(libusb_transfer *lut) class libusb_managed_recv_buffer_impl : public managed_recv_buffer { public: libusb_managed_recv_buffer_impl(libusb_transfer *lut, - usb_endpoint *endpoint) + usb_endpoint::sptr endpoint) : _buff(lut->buffer, lut->length) { _lut = lut; _endpoint = endpoint; } - ~libusb_managed_recv_buffer_impl() - { - if (!_endpoint->submit(_lut)) - std::cerr << "USB: failed to submit IN transfer" << std::endl; + ~libusb_managed_recv_buffer_impl(void){ + _endpoint->submit(_lut); } private: @@ -535,7 +311,7 @@ private: } libusb_transfer *_lut; - usb_endpoint *_endpoint; + usb_endpoint::sptr _endpoint; const boost::asio::const_buffer _buff; }; @@ -551,7 +327,7 @@ private: class libusb_managed_send_buffer_impl : public managed_send_buffer { public: libusb_managed_send_buffer_impl(libusb_transfer *lut, - usb_endpoint *endpoint, + usb_endpoint::sptr endpoint, size_t buff_size) : _buff(lut->buffer, buff_size), _committed(false) { @@ -559,8 +335,7 @@ public: _endpoint = endpoint; } - ~libusb_managed_send_buffer_impl() - { + ~libusb_managed_send_buffer_impl(void){ if (!_committed) { _lut->length = 0; _lut->actual_length = 0; @@ -580,12 +355,14 @@ public: _lut->length = num_bytes; _lut->actual_length = 0; - if (_endpoint->submit(_lut)) { + try{ + _endpoint->submit(_lut); _committed = true; return num_bytes; } - else { - return 0; + catch(const std::exception &e){ + std::cerr << "Error in commit: " << e.what() << std::endl; + return -1; } } @@ -596,7 +373,7 @@ private: } libusb_transfer *_lut; - usb_endpoint *_endpoint; + usb_endpoint::sptr _endpoint; const boost::asio::mutable_buffer _buff; bool _committed; }; @@ -608,11 +385,9 @@ private: class libusb_zero_copy_impl : public usb_zero_copy { private: - usb_endpoint *_rx_ep; - usb_endpoint *_tx_ep; + usb_endpoint::sptr _rx_ep, _tx_ep; libusb::device_handle::sptr _handle; - libusb::session::sptr _session; size_t _recv_buff_size; size_t _send_buff_size; @@ -626,8 +401,6 @@ public: unsigned int tx_endpoint, size_t recv_buff_size, size_t send_buff_size); - - ~libusb_zero_copy_impl(); managed_recv_buffer::sptr get_recv_buff(void); managed_send_buffer::sptr get_send_buff(void); @@ -646,45 +419,38 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(libusb::device_handle::sptr handle, unsigned int tx_endpoint, size_t buff_size, size_t block_size) - : _handle(handle), _session(libusb::session::get_global_session()), + : _handle(handle), _recv_buff_size(block_size), _send_buff_size(block_size), _num_frames(buff_size / block_size) { _handle->claim_interface(2 /*in interface*/); _handle->claim_interface(1 /*out interface*/); - _rx_ep = new usb_endpoint(_handle, // libusb device_handle - _session, // libusb session w/ context + _rx_ep = usb_endpoint::sptr(new usb_endpoint( + _handle, // libusb device_handle rx_endpoint, // USB endpoint number true, // IN endpoint _recv_buff_size, // buffer size per transfer - _num_frames); // number of libusb transfers + _num_frames // number of libusb transfers + )); - _tx_ep = new usb_endpoint(_handle, // libusb device_handle - _session, // libusb session w/ context + _tx_ep = usb_endpoint::sptr(new usb_endpoint( + _handle, // libusb device_handle tx_endpoint, // USB endpoint number false, // OUT endpoint _send_buff_size, // buffer size per transfer - _num_frames); // number of libusb transfers -} - - -libusb_zero_copy_impl::~libusb_zero_copy_impl() -{ - delete _rx_ep; - delete _tx_ep; + _num_frames // number of libusb transfers + )); } - /* * Construct a managed receive buffer from a completed libusb transfer * (happy with buffer full of data) obtained from the receive endpoint. * Return empty pointer if no transfer is available (timeout or error). * \return pointer to a managed receive buffer */ -managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff() -{ - libusb_transfer *lut = _rx_ep->get_completed_transfer(); +managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(void){ + libusb_transfer *lut = _rx_ep->get_lut_with_wait(/* TODO timeout API */); if (lut == NULL) { return managed_recv_buffer::sptr(); } @@ -702,9 +468,8 @@ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff() * (timeout or error). * \return pointer to a managed send buffer */ -managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff() -{ - libusb_transfer *lut = _tx_ep->get_free_transfer(); +managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(void){ + libusb_transfer *lut = _tx_ep->get_lut_with_wait(/* TODO timeout API */); if (lut == NULL) { return managed_send_buffer::sptr(); } @@ -736,6 +501,3 @@ usb_zero_copy::sptr usb_zero_copy::make(usb_device_handle::sptr handle, buff_size, block_size)); } - - - -- cgit v1.2.3 From 86213f8de2cc4ca671f9afbcf475c744fe7aebcb Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sun, 26 Sep 2010 20:54:58 -0700 Subject: usb: disable thread interruption on wait calls --- host/lib/transport/libusb1_zero_copy.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index e84793e88..d3888240b 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -132,6 +133,7 @@ static void callback(libusb_transfer *lut){ * \param pointer to libusb_transfer */ void usb_endpoint::callback_handle_transfer(libusb_transfer *lut){ + boost::this_thread::disable_interruption di; //disable because the wait can throw _completed_list->push_with_wait(lut); } @@ -182,10 +184,7 @@ usb_endpoint::~usb_endpoint(void){ } //collect canceled transfers (drain the queue) - libusb_transfer *lut; - while(_completed_list->pop_with_timed_wait( - lut, boost::posix_time::milliseconds(100) - )); + while (this->get_lut_with_wait() != NULL); //free all transfers BOOST_FOREACH(libusb_transfer *lut, _all_luts){ @@ -273,6 +272,7 @@ void usb_endpoint::print_transfer_status(libusb_transfer *lut){ } libusb_transfer *usb_endpoint::get_lut_with_wait(size_t timeout_ms){ + boost::this_thread::disable_interruption di; //disable because the wait can throw libusb_transfer *lut; if (_completed_list->pop_with_timed_wait( lut, boost::posix_time::milliseconds(timeout_ms) -- cgit v1.2.3 From 2f62c39b7135728ed06faa06db46004323b1c70f Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Mon, 27 Sep 2010 11:35:28 -0700 Subject: uhd: fix warning by adding brackets for while(cond){}; --- host/examples/test_async_messages.cpp | 2 +- host/lib/transport/libusb1_zero_copy.cpp | 2 +- host/lib/usrp/usrp2/io_impl.cpp | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/examples/test_async_messages.cpp b/host/examples/test_async_messages.cpp index 4c9d18121..0a5c076ea 100644 --- a/host/examples/test_async_messages.cpp +++ b/host/examples/test_async_messages.cpp @@ -58,7 +58,7 @@ void test_no_async_message(uhd::usrp::single_usrp::sptr sdev){ " Got unexpected event code 0x%x.\n" ) % async_md.event_code << std::endl; //clear the async messages - while (dev->recv_async_msg(async_md, 0)); + while (dev->recv_async_msg(async_md, 0)){}; } else{ std::cout << boost::format( diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index d3888240b..b089d11e0 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -184,7 +184,7 @@ usb_endpoint::~usb_endpoint(void){ } //collect canceled transfers (drain the queue) - while (this->get_lut_with_wait() != NULL); + while (this->get_lut_with_wait() != NULL){}; //free all transfers BOOST_FOREACH(libusb_transfer *lut, _all_luts){ diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp index 91a1b2344..65411801d 100644 --- a/host/lib/usrp/usrp2/io_impl.cpp +++ b/host/lib/usrp/usrp2/io_impl.cpp @@ -150,7 +150,7 @@ void usrp2_impl::io_init(void){ std::memcpy(send_buff->cast(), &data, sizeof(data)); send_buff->commit(sizeof(data)); //drain the recv buffers (may have junk) - while (data_transport->get_recv_buff().get()); + while (data_transport->get_recv_buff().get()){}; } //the number of recv frames is the number for the first transport -- cgit v1.2.3 From 543a63648f11d0e502e897f3cd98667005580c9e Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Wed, 29 Sep 2010 15:35:54 -0700 Subject: usrp: transfer resize options and documentation --- host/docs/usrp1.rst | 27 ++++++ host/include/uhd/transport/usb_zero_copy.hpp | 24 +++-- host/lib/transport/libusb1_zero_copy.cpp | 140 ++++++++++++++------------- host/lib/usrp/usrp1/usrp1_impl.cpp | 26 +++-- 4 files changed, 137 insertions(+), 80 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/docs/usrp1.rst b/host/docs/usrp1.rst index 3c1431d30..7cf447719 100644 --- a/host/docs/usrp1.rst +++ b/host/docs/usrp1.rst @@ -95,6 +95,27 @@ Notice that the subdevice name is always specified in the 3 possible cases. B:B + +------------------------------------------------------------------------ +Change USB transfer parameters +------------------------------------------------------------------------ +The advanced user may manipulate parameters of the usb bulk transfers +for various reasons, such as lowering latency or increasing buffer size. +By default, the UHD will use values for these parameters +that are known to perform well on a variety of systems. + +The following device address can be used to manipulate USB bulk transfers: + +* **recv_xfer_size:** the size of each receive bulk transfer in bytes +* **recv_num_xfers:** the number of simultaneous receive bulk transfers +* **send_xfer_size:** the size of each send bulk transfer in bytes +* **send_num_xfers:** the number of simultaneous send bulk transfers + +Example, set the args string to the following: +:: + + serial=12345678, recv_num_xfers=16 + ------------------------------------------------------------------------ OS Specific Notes ------------------------------------------------------------------------ @@ -113,3 +134,9 @@ so that non-root users may access the device: sudo mv tmpfile /etc/udev/rules.d/10-usrp.rules sudo udevadm control --reload-rules +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +Install libusb driver on Windows +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +On Windows, a driver must be installed the first time the USRP1 is attached to the host computer. +A download link for this driver can be found on the Ettus Research UHD wiki page. +Download and unpack the driver, and direct the Windows driver install wizard to the *.inf file. diff --git a/host/include/uhd/transport/usb_zero_copy.hpp b/host/include/uhd/transport/usb_zero_copy.hpp index 75232c22a..61bf380ba 100644 --- a/host/include/uhd/transport/usb_zero_copy.hpp +++ b/host/include/uhd/transport/usb_zero_copy.hpp @@ -45,16 +45,22 @@ public: * The underlying implementation may be platform specific. * * \param handle a device handle that uniquely identifying the device - * \param rx_endpoint an integer specifiying an IN endpoint number - * \param tx_endpoint an integer specifiying an OUT endpoint number - * \param buff_size total number of bytes of buffer space to allocate - * \param block_size number of bytes allocated for each I/O transaction + * \param recv_endpoint an integer specifiying an IN endpoint number + * \param send_endpoint an integer specifiying an OUT endpoint number + * \param recv_xfer_size the number of bytes for each receive transfer + * \param recv_num_xfers the number of simultaneous receive transfers + * \param send_xfer_size the number of bytes for each send transfer + * \param send_num_xfers the number of simultaneous send transfers */ - static sptr make(usb_device_handle::sptr handle, - unsigned int rx_endpoint, - unsigned int tx_endpoint, - size_t buff_size = 0, - size_t block_size = 0); + static sptr make( + usb_device_handle::sptr handle, + unsigned int recv_endpoint, + unsigned int send_endpoint, + size_t recv_xfer_size = 0, + size_t recv_num_xfers = 0, + size_t send_xfer_size = 0, + size_t send_num_xfers = 0 + ); }; }} //namespace diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index b089d11e0..f2dcff6b5 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -30,6 +30,9 @@ using namespace uhd::transport; const int libusb_timeout = 0; +static const size_t DEFAULT_NUM_XFERS = 16; //num xfers +static const size_t DEFAULT_XFER_SIZE = 32*512; //bytes + /*********************************************************************** * Helper functions ***********************************************************************/ @@ -56,7 +59,7 @@ void pp_transfer(libusb_transfer *lut) * create a bidirectional interface. It is a zero copy implementation * with respect to libusb, however, each send and recv requires a copy * operation from kernel to userspace; this is due to the usbfs - * interface provided by the kernel. + * interface provided by the kernel. **********************************************************************/ class usb_endpoint { public: @@ -107,7 +110,7 @@ private: //! a list of shared arrays for the transfer buffers std::vector > _buffers; - // Calls for processing asynchronous I/O + // Calls for processing asynchronous I/O libusb_transfer *allocate_transfer(int buff_len); void print_transfer_status(libusb_transfer *lut); }; @@ -142,7 +145,7 @@ void usb_endpoint::callback_handle_transfer(libusb_transfer *lut){ * Constructor * Allocate libusb transfers and mark as free. For IN endpoints, * submit the transfers so that they're ready to return when - * data is available. + * data is available. */ usb_endpoint::usb_endpoint( libusb::device_handle::sptr handle, @@ -194,7 +197,7 @@ usb_endpoint::~usb_endpoint(void){ /* - * Allocate a libusb transfer + * Allocate a libusb transfer * The allocated transfer - and buffer it contains - is repeatedly * submitted, reaped, and reused and should not be freed until shutdown. * \param buff_len size of the individual buffer held by each transfer @@ -225,7 +228,7 @@ libusb_transfer *usb_endpoint::allocate_transfer(int buff_len){ * Asynchonous transfer submission * Submit a libusb transfer to libusb add pending status * \param lut pointer to libusb_transfer - * \return true on success or false on error + * \return true on success or false on error */ void usb_endpoint::submit(libusb_transfer *lut){ UHD_ASSERT_THROW(libusb_submit_transfer(lut) == 0); @@ -281,14 +284,14 @@ libusb_transfer *usb_endpoint::get_lut_with_wait(size_t timeout_ms){ } /*********************************************************************** - * Managed buffers + * Managed buffers **********************************************************************/ /* * Libusb managed receive buffer * Construct a recv buffer from a libusb transfer. The memory held by * the libusb transfer is exposed through the managed buffer interface. * Upon destruction, the transfer and buffer are resubmitted to the - * endpoint for further use. + * endpoint for further use. */ class libusb_managed_recv_buffer_impl : public managed_recv_buffer { public: @@ -307,7 +310,7 @@ public: private: const boost::asio::const_buffer &get() const { - return _buff; + return _buff; } libusb_transfer *_lut; @@ -327,9 +330,8 @@ private: class libusb_managed_send_buffer_impl : public managed_send_buffer { public: libusb_managed_send_buffer_impl(libusb_transfer *lut, - usb_endpoint::sptr endpoint, - size_t buff_size) - : _buff(lut->buffer, buff_size), _committed(false) + usb_endpoint::sptr endpoint) + : _buff(lut->buffer, lut->length), _committed(false) { _lut = lut; _endpoint = endpoint; @@ -349,7 +351,7 @@ public: std::cerr << "UHD: send buffer already committed" << std::endl; return 0; } - + UHD_ASSERT_THROW(num_bytes <= boost::asio::buffer_size(_buff)); _lut->length = num_bytes; @@ -369,7 +371,7 @@ public: private: const boost::asio::mutable_buffer &get() const { - return _buff; + return _buff; } libusb_transfer *_lut; @@ -385,61 +387,71 @@ private: class libusb_zero_copy_impl : public usb_zero_copy { private: - usb_endpoint::sptr _rx_ep, _tx_ep; - libusb::device_handle::sptr _handle; - - size_t _recv_buff_size; - size_t _send_buff_size; - size_t _num_frames; + size_t _recv_num_frames, _send_num_frames; + usb_endpoint::sptr _recv_ep, _send_ep; public: typedef boost::shared_ptr sptr; - libusb_zero_copy_impl(libusb::device_handle::sptr handle, - unsigned int rx_endpoint, - unsigned int tx_endpoint, - size_t recv_buff_size, - size_t send_buff_size); + libusb_zero_copy_impl( + libusb::device_handle::sptr handle, + unsigned int recv_endpoint, unsigned int send_endpoint, + size_t recv_xfer_size, size_t recv_num_xfers, + size_t send_xfer_size, size_t send_num_xfers + ); managed_recv_buffer::sptr get_recv_buff(void); managed_send_buffer::sptr get_send_buff(void); - size_t get_num_recv_frames(void) const { return _num_frames; } - size_t get_num_send_frames(void) const { return _num_frames; } + size_t get_num_recv_frames(void) const { return _recv_num_frames; } + size_t get_num_send_frames(void) const { return _send_num_frames; } }; /* * Constructor * Initializes libusb, opens devices, and sets up interfaces for I/O. - * Finally, creates endpoints for asynchronous I/O. + * Finally, creates endpoints for asynchronous I/O. */ -libusb_zero_copy_impl::libusb_zero_copy_impl(libusb::device_handle::sptr handle, - unsigned int rx_endpoint, - unsigned int tx_endpoint, - size_t buff_size, - size_t block_size) - : _handle(handle), - _recv_buff_size(block_size), _send_buff_size(block_size), - _num_frames(buff_size / block_size) -{ - _handle->claim_interface(2 /*in interface*/); +libusb_zero_copy_impl::libusb_zero_copy_impl( + libusb::device_handle::sptr handle, + unsigned int recv_endpoint, unsigned int send_endpoint, + size_t recv_xfer_size, size_t recv_num_xfers, + size_t send_xfer_size, size_t send_num_xfers +){ + _handle = handle; + + //if the sizes are left at 0 (automatic) -> use the defaults + if (recv_xfer_size == 0) recv_xfer_size = DEFAULT_XFER_SIZE; + if (recv_num_xfers == 0) recv_num_xfers = DEFAULT_NUM_XFERS; + if (send_xfer_size == 0) send_xfer_size = DEFAULT_XFER_SIZE; + if (send_num_xfers == 0) send_num_xfers = DEFAULT_NUM_XFERS; + + //sanity check the transfer sizes + UHD_ASSERT_THROW(recv_xfer_size % 512 == 0); + UHD_ASSERT_THROW(send_xfer_size % 512 == 0); + + //store the num xfers for the num frames count + _recv_num_frames = recv_num_xfers; + _send_num_frames = send_num_xfers; + + _handle->claim_interface(2 /*in interface*/); _handle->claim_interface(1 /*out interface*/); - _rx_ep = usb_endpoint::sptr(new usb_endpoint( + _recv_ep = usb_endpoint::sptr(new usb_endpoint( _handle, // libusb device_handle - rx_endpoint, // USB endpoint number + recv_endpoint, // USB endpoint number true, // IN endpoint - _recv_buff_size, // buffer size per transfer - _num_frames // number of libusb transfers + recv_xfer_size, // buffer size per transfer + recv_num_xfers // number of libusb transfers )); - _tx_ep = usb_endpoint::sptr(new usb_endpoint( + _send_ep = usb_endpoint::sptr(new usb_endpoint( _handle, // libusb device_handle - tx_endpoint, // USB endpoint number + send_endpoint, // USB endpoint number false, // OUT endpoint - _send_buff_size, // buffer size per transfer - _num_frames // number of libusb transfers + send_xfer_size, // buffer size per transfer + send_num_xfers // number of libusb transfers )); } @@ -447,17 +459,17 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(libusb::device_handle::sptr handle, * Construct a managed receive buffer from a completed libusb transfer * (happy with buffer full of data) obtained from the receive endpoint. * Return empty pointer if no transfer is available (timeout or error). - * \return pointer to a managed receive buffer + * \return pointer to a managed receive buffer */ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(void){ - libusb_transfer *lut = _rx_ep->get_lut_with_wait(/* TODO timeout API */); + libusb_transfer *lut = _recv_ep->get_lut_with_wait(/* TODO timeout API */); if (lut == NULL) { return managed_recv_buffer::sptr(); } else { return managed_recv_buffer::sptr( new libusb_managed_recv_buffer_impl(lut, - _rx_ep)); + _recv_ep)); } } @@ -466,38 +478,36 @@ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(void){ * Construct a managed send buffer from a free libusb transfer (with * empty buffer). Return empty pointer of no transfer is available * (timeout or error). - * \return pointer to a managed send buffer + * \return pointer to a managed send buffer */ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(void){ - libusb_transfer *lut = _tx_ep->get_lut_with_wait(/* TODO timeout API */); + libusb_transfer *lut = _send_ep->get_lut_with_wait(/* TODO timeout API */); if (lut == NULL) { return managed_send_buffer::sptr(); } else { return managed_send_buffer::sptr( new libusb_managed_send_buffer_impl(lut, - _tx_ep, - _send_buff_size)); + _send_ep)); } } - /*********************************************************************** * USB zero_copy make functions **********************************************************************/ -usb_zero_copy::sptr usb_zero_copy::make(usb_device_handle::sptr handle, - unsigned int rx_endpoint, - unsigned int tx_endpoint, - size_t buff_size, - size_t block_size) - -{ +usb_zero_copy::sptr usb_zero_copy::make( + usb_device_handle::sptr handle, + unsigned int recv_endpoint, unsigned int send_endpoint, + size_t recv_xfer_size, size_t recv_num_xfers, + size_t send_xfer_size, size_t send_num_xfers +){ libusb::device_handle::sptr dev_handle(libusb::device_handle::get_cached_handle( boost::static_pointer_cast(handle)->get_device() )); - return sptr(new libusb_zero_copy_impl(dev_handle, - rx_endpoint, - tx_endpoint, - buff_size, - block_size)); + return sptr(new libusb_zero_copy_impl( + dev_handle, + recv_endpoint, send_endpoint, + recv_xfer_size, recv_num_xfers, + send_xfer_size, send_num_xfers + )); } diff --git a/host/lib/usrp/usrp1/usrp1_impl.cpp b/host/lib/usrp/usrp1/usrp1_impl.cpp index b4c23bf12..793e3027d 100644 --- a/host/lib/usrp/usrp1/usrp1_impl.cpp +++ b/host/lib/usrp/usrp1/usrp1_impl.cpp @@ -30,6 +30,7 @@ #include #include #include +#include #include using namespace uhd; @@ -107,6 +108,15 @@ static device_addrs_t usrp1_find(const device_addr_t &hint) /*********************************************************************** * Make **********************************************************************/ +template static output_type cast_from_dev_addr( + const device_addr_t &device_addr, + const std::string &key, + output_type def_val +){ + return (device_addr.has_key(key))? + boost::lexical_cast(device_addr[key]) : def_val; +} + static device::sptr usrp1_make(const device_addr_t &device_addr) { //extract the FPGA path for the USRP1 @@ -130,11 +140,15 @@ static device::sptr usrp1_make(const device_addr_t &device_addr) usrp_ctrl = usrp_ctrl::make(ctrl_transport); usrp_ctrl->usrp_load_fpga(usrp1_fpga_image); - data_transport = usb_zero_copy::make(handle, // identifier - 6, // IN endpoint - 2, // OUT endpoint - 2 * (1 << 20), // buffer size - 16384); // transfer size + data_transport = usb_zero_copy::make( + handle, // identifier + 6, // IN endpoint + 2, // OUT endpoint + size_t(cast_from_dev_addr(device_addr, "recv_xfer_size", 0)), + size_t(cast_from_dev_addr(device_addr, "recv_num_xfers", 0)), + size_t(cast_from_dev_addr(device_addr, "send_xfer_size", 0)), + size_t(cast_from_dev_addr(device_addr, "send_num_xfers", 0)) + ); break; } } @@ -173,7 +187,7 @@ usrp1_impl::usrp1_impl(uhd::transport::usb_zero_copy::sptr data_transport, //initialize the mboard mboard_init(); - //initialize the dboards + //initialize the dboards dboard_init(); //initialize the dsps -- cgit v1.2.3 From 2c8a7c7debf19d92065661cc1d258f97bd38e224 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Thu, 30 Sep 2010 14:36:24 -0700 Subject: uhd: implemented recv timeout for zero copy interface --- host/include/uhd/transport/zero_copy.hpp | 11 +++++--- host/lib/transport/libusb1_zero_copy.cpp | 42 +++++++++++++++---------------- host/lib/transport/udp_zero_copy_asio.cpp | 5 ++-- host/lib/transport/zero_copy.cpp | 4 +-- host/lib/usrp/usrp1/io_impl.cpp | 8 +++--- host/lib/usrp/usrp2/io_impl.cpp | 5 ++-- 6 files changed, 39 insertions(+), 36 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/include/uhd/transport/zero_copy.hpp b/host/include/uhd/transport/zero_copy.hpp index 513291b63..8ecafd3fb 100644 --- a/host/include/uhd/transport/zero_copy.hpp +++ b/host/include/uhd/transport/zero_copy.hpp @@ -122,9 +122,10 @@ namespace uhd{ namespace transport{ /*! * Get a new receive buffer from this transport object. + * \param timeout_ms the timeout to get the buffer in ms * \return a managed buffer, or null sptr on timeout/error */ - virtual managed_recv_buffer::sptr get_recv_buff(void) = 0; + virtual managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms) = 0; /*! * Get the maximum number of receive frames: @@ -171,16 +172,19 @@ namespace uhd{ namespace transport{ /*! * Get a new receive buffer from this transport object. + * \param timeout_ms the timeout to get the buffer in ms + * \return a managed buffer, or null sptr on timeout/error */ - managed_recv_buffer::sptr get_recv_buff(void); + managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms); private: /*! * Perform a private copying recv. * \param buff the buffer to write data into + * \param timeout_ms the timeout to get the buffer in ms * \return the number of bytes written to buff, 0 for timeout, negative for error */ - virtual ssize_t recv(const boost::asio::mutable_buffer &buff) = 0; + virtual ssize_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms) = 0; UHD_PIMPL_DECL(impl) _impl; }; @@ -204,6 +208,7 @@ namespace uhd{ namespace transport{ /*! * Get a new send buffer from this transport object. + * \return a managed buffer, or null sptr on timeout/error */ managed_send_buffer::sptr get_send_buff(void); diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index f2dcff6b5..f9beb0b4c 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -308,8 +308,7 @@ public: } private: - const boost::asio::const_buffer &get() const - { + const boost::asio::const_buffer &get(void) const{ return _buff; } @@ -369,8 +368,7 @@ public: } private: - const boost::asio::mutable_buffer &get() const - { + const boost::asio::mutable_buffer &get(void) const{ return _buff; } @@ -395,13 +393,13 @@ public: typedef boost::shared_ptr sptr; libusb_zero_copy_impl( - libusb::device_handle::sptr handle, - unsigned int recv_endpoint, unsigned int send_endpoint, - size_t recv_xfer_size, size_t recv_num_xfers, - size_t send_xfer_size, size_t send_num_xfers - ); + libusb::device_handle::sptr handle, + unsigned int recv_endpoint, unsigned int send_endpoint, + size_t recv_xfer_size, size_t recv_num_xfers, + size_t send_xfer_size, size_t send_num_xfers + ); - managed_recv_buffer::sptr get_recv_buff(void); + managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms); managed_send_buffer::sptr get_send_buff(void); size_t get_num_recv_frames(void) const { return _recv_num_frames; } @@ -419,23 +417,23 @@ libusb_zero_copy_impl::libusb_zero_copy_impl( size_t recv_xfer_size, size_t recv_num_xfers, size_t send_xfer_size, size_t send_num_xfers ){ - _handle = handle; + _handle = handle; - //if the sizes are left at 0 (automatic) -> use the defaults - if (recv_xfer_size == 0) recv_xfer_size = DEFAULT_XFER_SIZE; - if (recv_num_xfers == 0) recv_num_xfers = DEFAULT_NUM_XFERS; - if (send_xfer_size == 0) send_xfer_size = DEFAULT_XFER_SIZE; - if (send_num_xfers == 0) send_num_xfers = DEFAULT_NUM_XFERS; + //if the sizes are left at 0 (automatic) -> use the defaults + if (recv_xfer_size == 0) recv_xfer_size = DEFAULT_XFER_SIZE; + if (recv_num_xfers == 0) recv_num_xfers = DEFAULT_NUM_XFERS; + if (send_xfer_size == 0) send_xfer_size = DEFAULT_XFER_SIZE; + if (send_num_xfers == 0) send_num_xfers = DEFAULT_NUM_XFERS; //sanity check the transfer sizes UHD_ASSERT_THROW(recv_xfer_size % 512 == 0); UHD_ASSERT_THROW(send_xfer_size % 512 == 0); - //store the num xfers for the num frames count - _recv_num_frames = recv_num_xfers; - _send_num_frames = send_num_xfers; + //store the num xfers for the num frames count + _recv_num_frames = recv_num_xfers; + _send_num_frames = send_num_xfers; - _handle->claim_interface(2 /*in interface*/); + _handle->claim_interface(2 /*in interface*/); _handle->claim_interface(1 /*out interface*/); _recv_ep = usb_endpoint::sptr(new usb_endpoint( @@ -461,8 +459,8 @@ libusb_zero_copy_impl::libusb_zero_copy_impl( * Return empty pointer if no transfer is available (timeout or error). * \return pointer to a managed receive buffer */ -managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(void){ - libusb_transfer *lut = _recv_ep->get_lut_with_wait(/* TODO timeout API */); +managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(size_t timeout_ms){ + libusb_transfer *lut = _recv_ep->get_lut_with_wait(timeout_ms); if (lut == NULL) { return managed_recv_buffer::sptr(); } diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index ee989ee2b..0a6c9f2af 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -35,7 +35,6 @@ static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(sizeof(boost::uint32_t) * 2 //Perhaps this is due to the kernel scheduling, //but may change with host-based flow control. static const size_t MIN_SEND_SOCK_BUFF_SIZE = size_t(10e3); -static const double RECV_TIMEOUT = 0.1; //100 ms /*********************************************************************** * Zero Copy UDP implementation with ASIO: @@ -110,11 +109,11 @@ private: boost::asio::io_service _io_service; int _sock_fd; - ssize_t recv(const boost::asio::mutable_buffer &buff){ + ssize_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){ //setup timeval for timeout timeval tv; tv.tv_sec = 0; - tv.tv_usec = int(RECV_TIMEOUT*1e6); + tv.tv_usec = timeout_ms*1000; //setup rset for timeout fd_set rset; diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp index 8a1cde694..1fcf846a0 100644 --- a/host/lib/transport/zero_copy.cpp +++ b/host/lib/transport/zero_copy.cpp @@ -68,12 +68,12 @@ phony_zero_copy_recv_if::~phony_zero_copy_recv_if(void){ /* NOP */ } -managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(void){ +managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(size_t timeout_ms){ //allocate memory boost::uint8_t *recv_mem = new boost::uint8_t[_impl->max_buff_size]; //call recv() with timeout option - ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size)); + ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout_ms); if (num_bytes <= 0) return managed_recv_buffer::sptr(); //NULL sptr diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp index 73974f2d6..aee760a83 100644 --- a/host/lib/usrp/usrp1/io_impl.cpp +++ b/host/lib/usrp/usrp1/io_impl.cpp @@ -272,18 +272,18 @@ static void usrp1_bs_vrt_unpacker( } static bool get_recv_buffs( - zero_copy_if::sptr zc_if, + zero_copy_if::sptr zc_if, size_t timeout_ms, vrt_packet_handler::managed_recv_buffs_t &buffs ){ UHD_ASSERT_THROW(buffs.size() == 1); - buffs[0] = zc_if->get_recv_buff(); + buffs[0] = zc_if->get_recv_buff(timeout_ms); return buffs[0].get() != NULL; } size_t usrp1_impl::recv( const std::vector &buffs, size_t num_samps, rx_metadata_t &metadata, const io_type_t &io_type, - recv_mode_t recv_mode, size_t /*timeout_ms TODO*/ + recv_mode_t recv_mode, size_t timeout_ms ){ size_t num_samps_recvd = vrt_packet_handler::recv( _io_impl->packet_handler_recv_state, //last state of the recv handler @@ -292,7 +292,7 @@ size_t usrp1_impl::recv( io_type, _rx_otw_type, //input and output types to convert _clock_ctrl->get_master_clock_freq(), //master clock tick rate &usrp1_bs_vrt_unpacker, - boost::bind(&get_recv_buffs, _data_transport, _1), + boost::bind(&get_recv_buffs, _data_transport, timeout_ms, _1), &vrt_packet_handler::handle_overflow_nop, 0, //vrt header offset _rx_subdev_spec.size() //num channels diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp index 65411801d..3395f94e2 100644 --- a/host/lib/usrp/usrp2/io_impl.cpp +++ b/host/lib/usrp/usrp2/io_impl.cpp @@ -33,6 +33,7 @@ using namespace uhd::transport; namespace asio = boost::asio; static const int underflow_flags = async_metadata_t::EVENT_CODE_UNDERFLOW | async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET; +static const double RECV_TIMEOUT_MS = 100; /*********************************************************************** * io impl details (internal to this file) @@ -90,7 +91,7 @@ void usrp2_impl::io_impl::recv_pirate_loop( size_t next_packet_seq = 0; while(recv_pirate_crew_raiding){ - managed_recv_buffer::sptr buff = zc_if->get_recv_buff(); + managed_recv_buffer::sptr buff = zc_if->get_recv_buff(RECV_TIMEOUT_MS); if (not buff.get()) continue; //ignore timeout/error buffers try{ @@ -150,7 +151,7 @@ void usrp2_impl::io_init(void){ std::memcpy(send_buff->cast(), &data, sizeof(data)); send_buff->commit(sizeof(data)); //drain the recv buffers (may have junk) - while (data_transport->get_recv_buff().get()){}; + while (data_transport->get_recv_buff(RECV_TIMEOUT_MS).get()){}; } //the number of recv frames is the number for the first transport -- cgit v1.2.3 From 00cd6018405b57a0982b0ce103ff858c646ee18c Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Fri, 1 Oct 2010 18:22:41 -0700 Subject: uhd: implemented a double timeout (in seconds) for send and recv chains converted all size_t timeout_ms to double timeout bounded and alignment buffer now take double timeout added timeout to device::send and zero_copy_if::get_send_buff --- host/include/uhd/device.hpp | 27 ++++++++------ host/include/uhd/device.ipp | 9 +++-- host/include/uhd/transport/alignment_buffer.hpp | 9 ++--- host/include/uhd/transport/alignment_buffer.ipp | 15 +++----- host/include/uhd/transport/bounded_buffer.hpp | 12 ++---- host/include/uhd/transport/bounded_buffer.ipp | 15 ++++++-- host/include/uhd/transport/zero_copy.hpp | 18 +++++---- host/lib/transport/libusb1_zero_copy.cpp | 42 ++++++++++----------- host/lib/transport/udp_zero_copy_asio.cpp | 4 +- host/lib/transport/vrt_packet_handler.hpp | 49 +++++++++++++------------ host/lib/transport/zero_copy.cpp | 6 +-- host/lib/usrp/usrp1/io_impl.cpp | 20 +++++----- host/lib/usrp/usrp1/usrp1_impl.cpp | 4 +- host/lib/usrp/usrp1/usrp1_impl.hpp | 7 ++-- host/lib/usrp/usrp2/io_impl.cpp | 32 ++++++++-------- host/lib/usrp/usrp2/usrp2_impl.hpp | 6 +-- host/test/buffer_test.cpp | 2 +- 17 files changed, 140 insertions(+), 137 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/include/uhd/device.hpp b/host/include/uhd/device.hpp index 2077cae62..992276928 100644 --- a/host/include/uhd/device.hpp +++ b/host/include/uhd/device.hpp @@ -41,9 +41,6 @@ public: typedef boost::function find_t; typedef boost::function make_t; - //! A reasonable default timeout for receive - static const size_t default_recv_timeout_ms = 100; - /*! * Register a device into the discovery and factory system. * @@ -112,12 +109,15 @@ public: * * This is a blocking call and will not return until the number * of samples returned have been read out of each buffer. + * Under a timeout condition, the number of samples returned + * may be less than the number of samples specified. * * \param buffs a vector of read-only memory containing IF data * \param nsamps_per_buff the number of samples to send, per buffer * \param metadata data describing the buffer's contents * \param io_type the type of data loaded in the buffer * \param send_mode tells send how to unload the buffer + * \param timeout the timeout in seconds to wait on a packet * \return the number of samples sent */ virtual size_t send( @@ -125,7 +125,8 @@ public: size_t nsamps_per_buff, const tx_metadata_t &metadata, const io_type_t &io_type, - send_mode_t send_mode + send_mode_t send_mode, + double timeout = 0.1 ) = 0; /*! @@ -136,7 +137,8 @@ public: size_t nsamps_per_buff, const tx_metadata_t &metadata, const io_type_t &io_type, - send_mode_t send_mode + send_mode_t send_mode, + double timeout = 0.1 ); /*! @@ -154,7 +156,9 @@ public: * See the rx metadata fragment flags and offset fields for details. * * This is a blocking call and will not return until the number - * of samples returned have been written into each buffer or timeout. + * of samples returned have been written into each buffer. + * Under a timeout condition, the number of samples returned + * may be less than the number of samples specified. * * When using the full buffer recv mode, the metadata only applies * to the first packet received and written into the recv buffers. @@ -165,7 +169,7 @@ public: * \param metadata data to fill describing the buffer * \param io_type the type of data to fill into the buffer * \param recv_mode tells recv how to load the buffer - * \param timeout_ms the timeout in milliseconds to wait for a packet + * \param timeout the timeout in seconds to wait for a packet * \return the number of samples received or 0 on error */ virtual size_t recv( @@ -174,7 +178,7 @@ public: rx_metadata_t &metadata, const io_type_t &io_type, recv_mode_t recv_mode, - size_t timeout_ms = default_recv_timeout_ms + double timeout = 0.1 ) = 0; /*! @@ -186,7 +190,7 @@ public: rx_metadata_t &metadata, const io_type_t &io_type, recv_mode_t recv_mode, - size_t timeout_ms = default_recv_timeout_ms + double timeout = 0.1 ); /*! @@ -204,12 +208,11 @@ public: /*! * Receive and asynchronous message from the device. * \param async_metadata the metadata to be filled in - * \param timeout_ms the timeout in milliseconds to wait for a message + * \param timeout the timeout in seconds to wait for a message * \return true when the async_metadata is valid, false for timeout */ virtual bool recv_async_msg( - async_metadata_t &async_metadata, - size_t timeout_ms = default_recv_timeout_ms + async_metadata_t &async_metadata, double timeout = 0.1 ) = 0; }; diff --git a/host/include/uhd/device.ipp b/host/include/uhd/device.ipp index 60a3f535d..e2e51ecd0 100644 --- a/host/include/uhd/device.ipp +++ b/host/include/uhd/device.ipp @@ -25,12 +25,13 @@ namespace uhd{ size_t nsamps_per_buff, const tx_metadata_t &metadata, const io_type_t &io_type, - send_mode_t send_mode + send_mode_t send_mode, + double timeout ){ return this->send( std::vector(1, buff), nsamps_per_buff, metadata, - io_type, send_mode + io_type, send_mode, timeout ); } @@ -40,12 +41,12 @@ namespace uhd{ rx_metadata_t &metadata, const io_type_t &io_type, recv_mode_t recv_mode, - size_t timeout_ms + double timeout ){ return this->recv( std::vector(1, buff), nsamps_per_buff, metadata, - io_type, recv_mode, timeout_ms + io_type, recv_mode, timeout ); } diff --git a/host/include/uhd/transport/alignment_buffer.hpp b/host/include/uhd/transport/alignment_buffer.hpp index 29ba74efc..f44a037f8 100644 --- a/host/include/uhd/transport/alignment_buffer.hpp +++ b/host/include/uhd/transport/alignment_buffer.hpp @@ -48,20 +48,17 @@ namespace uhd{ namespace transport{ * \return true if the element fit without popping for space */ virtual bool push_with_pop_on_full( - const elem_type &elem, - const seq_type &seq, - size_t index + const elem_type &elem, const seq_type &seq, size_t index ) = 0; /*! * Pop an aligned set of elements from this alignment buffer. * \param elems a collection to store the aligned elements - * \param time the timeout time + * \param timeout the timeout in seconds * \return false when the operation times out */ virtual bool pop_elems_with_timed_wait( - std::vector &elems, - const time_duration_t &time + std::vector &elems, double timeout ) = 0; }; diff --git a/host/include/uhd/transport/alignment_buffer.ipp b/host/include/uhd/transport/alignment_buffer.ipp index 61b3b60f5..5f09de0d9 100644 --- a/host/include/uhd/transport/alignment_buffer.ipp +++ b/host/include/uhd/transport/alignment_buffer.ipp @@ -41,9 +41,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ } UHD_INLINE bool push_with_pop_on_full( - const elem_type &elem, - const seq_type &seq, - size_t index + const elem_type &elem, const seq_type &seq, size_t index ){ //clear the buffer for this index if the seqs are mis-ordered if (seq < _last_seqs[index]){ @@ -54,17 +52,16 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ } UHD_INLINE bool pop_elems_with_timed_wait( - std::vector &elems, - const time_duration_t &time + std::vector &elems, double timeout ){ - boost::system_time exit_time = boost::get_system_time() + time; + boost::system_time exit_time = boost::get_system_time() + boost::posix_time::microseconds(long(timeout*1e6)); buff_contents_type buff_contents_tmp; std::list 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, exit_time - boost::get_system_time() + buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() )) return false; elems[index] = buff_contents_tmp.first; seq_type expected_seq_id = buff_contents_tmp.second; @@ -79,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, exit_time - boost::get_system_time() + buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() )) return false; elems[index] = buff_contents_tmp.first; expected_seq_id = buff_contents_tmp.second; @@ -89,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, exit_time - boost::get_system_time() + buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() )) return false; //if the sequence id matches: diff --git a/host/include/uhd/transport/bounded_buffer.hpp b/host/include/uhd/transport/bounded_buffer.hpp index d1deece96..aca93b071 100644 --- a/host/include/uhd/transport/bounded_buffer.hpp +++ b/host/include/uhd/transport/bounded_buffer.hpp @@ -20,13 +20,9 @@ #include #include -#include namespace uhd{ namespace transport{ - //! typedef for the time duration type for wait operations - typedef boost::posix_time::time_duration time_duration_t; - /*! * Implement a templated bounded buffer: * Used for passing elements between threads in a producer-consumer model. @@ -64,10 +60,10 @@ namespace uhd{ namespace transport{ * Push a new element into the bounded_buffer. * Wait until the bounded_buffer becomes non-full or timeout. * \param elem the new element to push - * \param time the timeout time + * \param timeout the timeout in seconds * \return false when the operation times out */ - virtual bool push_with_timed_wait(const elem_type &elem, const time_duration_t &time) = 0; + virtual bool push_with_timed_wait(const elem_type &elem, double timeout) = 0; /*! * Pop an element from the bounded_buffer. @@ -80,10 +76,10 @@ namespace uhd{ namespace transport{ * Pop an element from the bounded_buffer. * Wait until the bounded_buffer becomes non-empty or timeout. * \param elem the element reference pop to - * \param time the timeout time + * \param timeout the timeout in seconds * \return false when the operation times out */ - virtual bool pop_with_timed_wait(elem_type &elem, const time_duration_t &time) = 0; + virtual bool pop_with_timed_wait(elem_type &elem, double timeout) = 0; /*! * Clear all elements from the bounded_buffer. diff --git a/host/include/uhd/transport/bounded_buffer.ipp b/host/include/uhd/transport/bounded_buffer.ipp index e106e229e..71143741e 100644 --- a/host/include/uhd/transport/bounded_buffer.ipp +++ b/host/include/uhd/transport/bounded_buffer.ipp @@ -21,6 +21,7 @@ #include #include #include +#include namespace uhd{ namespace transport{ namespace{ /*anon*/ @@ -57,9 +58,12 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ _empty_cond.notify_one(); } - bool push_with_timed_wait(const elem_type &elem, const time_duration_t &time){ + bool push_with_timed_wait(const elem_type &elem, double timeout){ boost::unique_lock lock(_mutex); - if (not _full_cond.timed_wait(lock, time, boost::bind(&bounded_buffer_impl::not_full, this))) return false; + if (not _full_cond.timed_wait( + lock, boost::posix_time::microseconds(long(timeout*1e6)), + boost::bind(&bounded_buffer_impl::not_full, this) + )) return false; _buffer.push_front(elem); lock.unlock(); _empty_cond.notify_one(); @@ -74,9 +78,12 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ _full_cond.notify_one(); } - bool pop_with_timed_wait(elem_type &elem, const time_duration_t &time){ + bool pop_with_timed_wait(elem_type &elem, double timeout){ boost::unique_lock lock(_mutex); - if (not _empty_cond.timed_wait(lock, time, boost::bind(&bounded_buffer_impl::not_empty, this))) return false; + if (not _empty_cond.timed_wait( + lock, boost::posix_time::microseconds(long(timeout*1e6)), + boost::bind(&bounded_buffer_impl::not_empty, this) + )) return false; elem = _buffer.back(); _buffer.pop_back(); lock.unlock(); _full_cond.notify_one(); diff --git a/host/include/uhd/transport/zero_copy.hpp b/host/include/uhd/transport/zero_copy.hpp index 8ecafd3fb..ba19b193c 100644 --- a/host/include/uhd/transport/zero_copy.hpp +++ b/host/include/uhd/transport/zero_copy.hpp @@ -122,10 +122,10 @@ namespace uhd{ namespace transport{ /*! * Get a new receive buffer from this transport object. - * \param timeout_ms the timeout to get the buffer in ms + * \param timeout the timeout to get the buffer in seconds * \return a managed buffer, or null sptr on timeout/error */ - virtual managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms) = 0; + virtual managed_recv_buffer::sptr get_recv_buff(double timeout = 0.1) = 0; /*! * Get the maximum number of receive frames: @@ -138,9 +138,10 @@ namespace uhd{ namespace transport{ /*! * Get a new send buffer from this transport object. + * \param timeout the timeout to get the buffer in seconds * \return a managed buffer, or null sptr on timeout/error */ - virtual managed_send_buffer::sptr get_send_buff(void) = 0; + virtual managed_send_buffer::sptr get_send_buff(double timeout = 0.1) = 0; /*! * Get the maximum number of send frames: @@ -172,19 +173,19 @@ namespace uhd{ namespace transport{ /*! * Get a new receive buffer from this transport object. - * \param timeout_ms the timeout to get the buffer in ms + * \param timeout the timeout to get the buffer in seconds * \return a managed buffer, or null sptr on timeout/error */ - managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms); + managed_recv_buffer::sptr get_recv_buff(double timeout); private: /*! * Perform a private copying recv. * \param buff the buffer to write data into - * \param timeout_ms the timeout to get the buffer in ms + * \param timeout the timeout to get the buffer in seconds * \return the number of bytes written to buff, 0 for timeout, negative for error */ - virtual ssize_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms) = 0; + virtual ssize_t recv(const boost::asio::mutable_buffer &buff, double timeout) = 0; UHD_PIMPL_DECL(impl) _impl; }; @@ -208,9 +209,10 @@ namespace uhd{ namespace transport{ /*! * Get a new send buffer from this transport object. + * \param timeout the timeout to get the buffer in seconds * \return a managed buffer, or null sptr on timeout/error */ - managed_send_buffer::sptr get_send_buff(void); + managed_send_buffer::sptr get_send_buff(double timeout); private: /*! diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index f9beb0b4c..7f2bc3468 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -24,12 +24,10 @@ #include #include #include -#include using namespace uhd::transport; -const int libusb_timeout = 0; - +static const double CLEANUP_TIMEOUT = 0.2; //seconds static const size_t DEFAULT_NUM_XFERS = 16; //num xfers static const size_t DEFAULT_XFER_SIZE = 32*512; //bytes @@ -84,10 +82,10 @@ public: * Get an available transfer: * For inputs, this is a just filled transfer. * For outputs, this is a just emptied transfer. - * \param timeout_ms the timeout to wait for a lut + * \param timeout the timeout to wait for a lut * \return the transfer pointer or NULL if timeout */ - libusb_transfer *get_lut_with_wait(size_t timeout_ms = 100); + libusb_transfer *get_lut_with_wait(double timeout); //Callback use only void callback_handle_transfer(libusb_transfer *lut); @@ -187,7 +185,7 @@ usb_endpoint::~usb_endpoint(void){ } //collect canceled transfers (drain the queue) - while (this->get_lut_with_wait() != NULL){}; + while (this->get_lut_with_wait(CLEANUP_TIMEOUT) != NULL){}; //free all transfers BOOST_FOREACH(libusb_transfer *lut, _all_luts){ @@ -274,12 +272,10 @@ void usb_endpoint::print_transfer_status(libusb_transfer *lut){ } } -libusb_transfer *usb_endpoint::get_lut_with_wait(size_t timeout_ms){ +libusb_transfer *usb_endpoint::get_lut_with_wait(double timeout){ boost::this_thread::disable_interruption di; //disable because the wait can throw libusb_transfer *lut; - if (_completed_list->pop_with_timed_wait( - lut, boost::posix_time::milliseconds(timeout_ms) - )) return lut; + if (_completed_list->pop_with_timed_wait(lut, timeout)) return lut; return NULL; } @@ -399,8 +395,8 @@ public: size_t send_xfer_size, size_t send_num_xfers ); - managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms); - managed_send_buffer::sptr get_send_buff(void); + managed_recv_buffer::sptr get_recv_buff(double); + managed_send_buffer::sptr get_send_buff(double); size_t get_num_recv_frames(void) const { return _recv_num_frames; } size_t get_num_send_frames(void) const { return _send_num_frames; } @@ -459,8 +455,8 @@ libusb_zero_copy_impl::libusb_zero_copy_impl( * Return empty pointer if no transfer is available (timeout or error). * \return pointer to a managed receive buffer */ -managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(size_t timeout_ms){ - libusb_transfer *lut = _recv_ep->get_lut_with_wait(timeout_ms); +managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(double timeout){ + libusb_transfer *lut = _recv_ep->get_lut_with_wait(timeout); if (lut == NULL) { return managed_recv_buffer::sptr(); } @@ -478,8 +474,8 @@ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(size_t timeout_ms * (timeout or error). * \return pointer to a managed send buffer */ -managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(void){ - libusb_transfer *lut = _send_ep->get_lut_with_wait(/* TODO timeout API */); +managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){ + libusb_transfer *lut = _send_ep->get_lut_with_wait(timeout); if (lut == NULL) { return managed_send_buffer::sptr(); } @@ -494,18 +490,18 @@ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(void){ * USB zero_copy make functions **********************************************************************/ usb_zero_copy::sptr usb_zero_copy::make( - usb_device_handle::sptr handle, + usb_device_handle::sptr handle, unsigned int recv_endpoint, unsigned int send_endpoint, - size_t recv_xfer_size, size_t recv_num_xfers, - size_t send_xfer_size, size_t send_num_xfers + size_t recv_xfer_size, size_t recv_num_xfers, + size_t send_xfer_size, size_t send_num_xfers ){ libusb::device_handle::sptr dev_handle(libusb::device_handle::get_cached_handle( boost::static_pointer_cast(handle)->get_device() )); return sptr(new libusb_zero_copy_impl( - dev_handle, - recv_endpoint, send_endpoint, - recv_xfer_size, recv_num_xfers, - send_xfer_size, send_num_xfers + dev_handle, + recv_endpoint, send_endpoint, + recv_xfer_size, recv_num_xfers, + send_xfer_size, send_num_xfers )); } diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 0a6c9f2af..3130830a5 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -109,11 +109,11 @@ private: boost::asio::io_service _io_service; int _sock_fd; - ssize_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){ + ssize_t recv(const boost::asio::mutable_buffer &buff, double timeout){ //setup timeval for timeout timeval tv; tv.tv_sec = 0; - tv.tv_usec = timeout_ms*1000; + tv.tv_usec = long(timeout*1e6); //setup rset for timeout fd_set rset; diff --git a/host/lib/transport/vrt_packet_handler.hpp b/host/lib/transport/vrt_packet_handler.hpp index b603f1371..e11afff30 100644 --- a/host/lib/transport/vrt_packet_handler.hpp +++ b/host/lib/transport/vrt_packet_handler.hpp @@ -303,18 +303,18 @@ template UHD_INLINE T get_context_code( * Pack a vrt header, copy-convert the data, and send it. * - helper function for vrt_packet_handler::send ******************************************************************/ - static UHD_INLINE void _send1( + static UHD_INLINE size_t _send1( send_state &state, const std::vector &buffs, - size_t offset_bytes, - size_t num_samps, + const size_t offset_bytes, + const size_t num_samps, uhd::transport::vrt::if_packet_info_t &if_packet_info, const uhd::io_type_t &io_type, const uhd::otw_type_t &otw_type, const vrt_packer_t &vrt_packer, const get_send_buffs_t &get_send_buffs, - size_t vrt_header_offset_words32, - size_t chans_per_otw_buff + const size_t vrt_header_offset_words32, + const size_t chans_per_otw_buff ){ //load the rest of the if_packet_info in here if_packet_info.num_payload_words32 = (num_samps*chans_per_otw_buff*otw_type.get_sample_size())/sizeof(boost::uint32_t); @@ -322,7 +322,7 @@ template UHD_INLINE T get_context_code( //get send buffers for each channel managed_send_buffs_t send_buffs(buffs.size()/chans_per_otw_buff); - UHD_ASSERT_THROW(get_send_buffs(send_buffs)); + if (not get_send_buffs(send_buffs)) return 0; std::vector io_buffs(chans_per_otw_buff); for (size_t i = 0; i < buffs.size(); i+=chans_per_otw_buff){ @@ -347,6 +347,7 @@ template UHD_INLINE T get_context_code( std::cerr << "commit to send buffer returned less than commit size" << std::endl; } } + return num_samps; } /******************************************************************* @@ -381,7 +382,6 @@ template UHD_INLINE T get_context_code( //////////////////////////////////////////////////////////////// case uhd::device::SEND_MODE_ONE_PACKET:{ //////////////////////////////////////////////////////////////// - size_t num_samps = std::min(total_num_samps, max_samples_per_packet); //fill in parts of the packet info overwrote in full buff mode if_packet_info.has_tsi = metadata.has_time_spec; @@ -389,10 +389,10 @@ template UHD_INLINE T get_context_code( if_packet_info.sob = metadata.start_of_burst; if_packet_info.eob = metadata.end_of_burst; - _send1( + return _send1( state, buffs, 0, - num_samps, + std::min(total_num_samps, max_samples_per_packet), if_packet_info, io_type, otw_type, vrt_packer, @@ -400,31 +400,32 @@ template UHD_INLINE T get_context_code( vrt_header_offset_words32, chans_per_otw_buff ); - return num_samps; } //////////////////////////////////////////////////////////////// case uhd::device::SEND_MODE_FULL_BUFF:{ //////////////////////////////////////////////////////////////// - //calculate constants for fragmentation - const size_t num_fragments = (total_num_samps+max_samples_per_packet-1)/max_samples_per_packet; - static const size_t first_fragment_index = 0; - const size_t final_fragment_index = num_fragments-1; + size_t total_num_samps_sent = 0; //loop through the following fragment indexes - for (size_t n = first_fragment_index; n <= final_fragment_index; n++){ + while(total_num_samps_sent < total_num_samps){ + + //calculate per-loop-iteration variables + const size_t total_num_samps_unsent = total_num_samps - total_num_samps_sent; + const bool first_fragment = (total_num_samps_sent == 0); + const bool final_fragment = (total_num_samps_unsent <= max_samples_per_packet); //calculate new flags for the fragments - if_packet_info.has_tsi = metadata.has_time_spec and (n == first_fragment_index); - if_packet_info.has_tsf = metadata.has_time_spec and (n == first_fragment_index); - if_packet_info.sob = metadata.start_of_burst and (n == first_fragment_index); - if_packet_info.eob = metadata.end_of_burst and (n == final_fragment_index); + if_packet_info.has_tsi = metadata.has_time_spec and first_fragment; + if_packet_info.has_tsf = if_packet_info.has_tsi; + if_packet_info.sob = metadata.start_of_burst and first_fragment; + if_packet_info.eob = metadata.end_of_burst and final_fragment; //send the fragment with the helper function - _send1( + const size_t num_samps_sent = _send1( state, - buffs, n*max_samples_per_packet*io_type.size, - (n == final_fragment_index)?(total_num_samps%max_samples_per_packet):max_samples_per_packet, + buffs, total_num_samps_sent*io_type.size, + std::min(total_num_samps_unsent, max_samples_per_packet), if_packet_info, io_type, otw_type, vrt_packer, @@ -432,8 +433,10 @@ template UHD_INLINE T get_context_code( vrt_header_offset_words32, chans_per_otw_buff ); + total_num_samps_sent += num_samps_sent; + if (num_samps_sent == 0) return total_num_samps_sent; } - return total_num_samps; + return total_num_samps_sent; } default: throw std::runtime_error("unknown send mode"); diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp index 1fcf846a0..dfb65951f 100644 --- a/host/lib/transport/zero_copy.cpp +++ b/host/lib/transport/zero_copy.cpp @@ -68,12 +68,12 @@ phony_zero_copy_recv_if::~phony_zero_copy_recv_if(void){ /* NOP */ } -managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(size_t timeout_ms){ +managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(double timeout){ //allocate memory boost::uint8_t *recv_mem = new boost::uint8_t[_impl->max_buff_size]; //call recv() with timeout option - ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout_ms); + ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout); if (num_bytes <= 0) return managed_recv_buffer::sptr(); //NULL sptr @@ -138,6 +138,6 @@ phony_zero_copy_send_if::~phony_zero_copy_send_if(void){ delete [] _impl->send_mem; } -managed_send_buffer::sptr phony_zero_copy_send_if::get_send_buff(void){ +managed_send_buffer::sptr phony_zero_copy_send_if::get_send_buff(double){ return _impl->send_buff; //FIXME there is only ever one send buff, we assume that the caller doesnt hang onto these } diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp index aee760a83..8d9c68961 100644 --- a/host/lib/usrp/usrp1/io_impl.cpp +++ b/host/lib/usrp/usrp1/io_impl.cpp @@ -94,12 +94,13 @@ struct usrp1_impl::io_impl{ //all of this to ensure only full buffers are committed managed_send_buffer::sptr send_buff; size_t num_bytes_committed; + double send_timeout; boost::uint8_t pseudo_buff[BYTES_PER_PACKET]; ssize_t phony_commit_pseudo_buff(size_t num_bytes); ssize_t phony_commit_send_buff(size_t num_bytes); ssize_t commit_send_buff(void); void flush_send_buff(void); - bool get_send_buffs(vrt_packet_handler::managed_send_buffs_t &); + bool get_send_buffs(vrt_packet_handler::managed_send_buffs_t &, double); //helpers to get at the send buffer + offset inline void *get_send_mem_ptr(void){ @@ -159,14 +160,15 @@ void usrp1_impl::io_impl::flush_send_buff(void){ */ ssize_t usrp1_impl::io_impl::commit_send_buff(void){ ssize_t ret = send_buff->commit(num_bytes_committed); - send_buff = data_transport->get_send_buff(); + send_buff = data_transport->get_send_buff(send_timeout); num_bytes_committed = 0; return ret; } bool usrp1_impl::io_impl::get_send_buffs( - vrt_packet_handler::managed_send_buffs_t &buffs + vrt_packet_handler::managed_send_buffs_t &buffs, double timeout ){ + send_timeout = timeout; UHD_ASSERT_THROW(buffs.size() == 1); //not enough bytes free -> use the pseudo buffer @@ -216,7 +218,7 @@ static void usrp1_bs_vrt_packer( size_t usrp1_impl::send( const std::vector &buffs, size_t num_samps, const tx_metadata_t &metadata, const io_type_t &io_type, - send_mode_t send_mode + send_mode_t send_mode, double timeout ){ size_t num_samps_sent = vrt_packet_handler::send( _io_impl->packet_handler_send_state, //last state of the send handler @@ -225,7 +227,7 @@ size_t usrp1_impl::send( io_type, _tx_otw_type, //input and output types to convert _clock_ctrl->get_master_clock_freq(), //master clock tick rate &usrp1_bs_vrt_packer, - boost::bind(&usrp1_impl::io_impl::get_send_buffs, _io_impl.get(), _1), + boost::bind(&usrp1_impl::io_impl::get_send_buffs, _io_impl.get(), _1, timeout), get_max_send_samps_per_packet(), 0, //vrt header offset _tx_subdev_spec.size() //num channels @@ -272,18 +274,18 @@ static void usrp1_bs_vrt_unpacker( } static bool get_recv_buffs( - zero_copy_if::sptr zc_if, size_t timeout_ms, + zero_copy_if::sptr zc_if, double timeout, vrt_packet_handler::managed_recv_buffs_t &buffs ){ UHD_ASSERT_THROW(buffs.size() == 1); - buffs[0] = zc_if->get_recv_buff(timeout_ms); + buffs[0] = zc_if->get_recv_buff(timeout); return buffs[0].get() != NULL; } size_t usrp1_impl::recv( const std::vector &buffs, size_t num_samps, rx_metadata_t &metadata, const io_type_t &io_type, - recv_mode_t recv_mode, size_t timeout_ms + recv_mode_t recv_mode, double timeout ){ size_t num_samps_recvd = vrt_packet_handler::recv( _io_impl->packet_handler_recv_state, //last state of the recv handler @@ -292,7 +294,7 @@ size_t usrp1_impl::recv( io_type, _rx_otw_type, //input and output types to convert _clock_ctrl->get_master_clock_freq(), //master clock tick rate &usrp1_bs_vrt_unpacker, - boost::bind(&get_recv_buffs, _data_transport, timeout_ms, _1), + boost::bind(&get_recv_buffs, _data_transport, timeout, _1), &vrt_packet_handler::handle_overflow_nop, 0, //vrt header offset _rx_subdev_spec.size() //num channels diff --git a/host/lib/usrp/usrp1/usrp1_impl.cpp b/host/lib/usrp/usrp1/usrp1_impl.cpp index 156fc119f..6ebd6bb09 100644 --- a/host/lib/usrp/usrp1/usrp1_impl.cpp +++ b/host/lib/usrp/usrp1/usrp1_impl.cpp @@ -209,9 +209,9 @@ usrp1_impl::~usrp1_impl(void){ /* NOP */ } -bool usrp1_impl::recv_async_msg(uhd::async_metadata_t &, size_t timeout_ms){ +bool usrp1_impl::recv_async_msg(uhd::async_metadata_t &, double timeout){ //dummy fill-in for the recv_async_msg - boost::this_thread::sleep(boost::posix_time::milliseconds(timeout_ms)); + boost::this_thread::sleep(boost::posix_time::microseconds(long(timeout*1e6))); return false; } diff --git a/host/lib/usrp/usrp1/usrp1_impl.hpp b/host/lib/usrp/usrp1/usrp1_impl.hpp index 20ae3c02a..f2c464610 100644 --- a/host/lib/usrp/usrp1/usrp1_impl.hpp +++ b/host/lib/usrp/usrp1/usrp1_impl.hpp @@ -83,13 +83,12 @@ public: size_t, const uhd::tx_metadata_t &, const uhd::io_type_t &, - send_mode_t); + send_mode_t, double); size_t recv(const std::vector &, size_t, uhd::rx_metadata_t &, const uhd::io_type_t &, - recv_mode_t, - size_t timeout); + recv_mode_t, double); static const size_t BYTES_PER_PACKET = 512*4; //under the transfer size @@ -101,7 +100,7 @@ public: return BYTES_PER_PACKET/_rx_otw_type.get_sample_size()/_rx_subdev_spec.size(); } - bool recv_async_msg(uhd::async_metadata_t &, size_t); + bool recv_async_msg(uhd::async_metadata_t &, double); private: /*! diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp index 3395f94e2..c0d8ab029 100644 --- a/host/lib/usrp/usrp2/io_impl.cpp +++ b/host/lib/usrp/usrp2/io_impl.cpp @@ -33,7 +33,6 @@ using namespace uhd::transport; namespace asio = boost::asio; static const int underflow_flags = async_metadata_t::EVENT_CODE_UNDERFLOW | async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET; -static const double RECV_TIMEOUT_MS = 100; /*********************************************************************** * io impl details (internal to this file) @@ -59,9 +58,9 @@ struct usrp2_impl::io_impl{ recv_pirate_crew.join_all(); } - bool get_recv_buffs(vrt_packet_handler::managed_recv_buffs_t &buffs, size_t timeout_ms){ + bool get_recv_buffs(vrt_packet_handler::managed_recv_buffs_t &buffs, double timeout){ boost::this_thread::disable_interruption di; //disable because the wait can throw - return recv_pirate_booty->pop_elems_with_timed_wait(buffs, boost::posix_time::milliseconds(timeout_ms)); + return recv_pirate_booty->pop_elems_with_timed_wait(buffs, timeout); } //state management for the vrt packet handler code @@ -91,7 +90,7 @@ void usrp2_impl::io_impl::recv_pirate_loop( size_t next_packet_seq = 0; while(recv_pirate_crew_raiding){ - managed_recv_buffer::sptr buff = zc_if->get_recv_buff(RECV_TIMEOUT_MS); + managed_recv_buffer::sptr buff = zc_if->get_recv_buff(); if (not buff.get()) continue; //ignore timeout/error buffers try{ @@ -151,7 +150,7 @@ void usrp2_impl::io_init(void){ std::memcpy(send_buff->cast(), &data, sizeof(data)); send_buff->commit(sizeof(data)); //drain the recv buffers (may have junk) - while (data_transport->get_recv_buff(RECV_TIMEOUT_MS).get()){}; + while (data_transport->get_recv_buff().get()){}; } //the number of recv frames is the number for the first transport @@ -179,12 +178,10 @@ void usrp2_impl::io_init(void){ * Async Data **********************************************************************/ bool usrp2_impl::recv_async_msg( - async_metadata_t &async_metadata, size_t timeout_ms + 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, boost::posix_time::milliseconds(timeout_ms) - ); + return _io_impl->async_msg_fifo->pop_with_timed_wait(async_metadata, timeout); } /*********************************************************************** @@ -192,19 +189,22 @@ bool usrp2_impl::recv_async_msg( **********************************************************************/ static bool get_send_buffs( const std::vector &trans, - vrt_packet_handler::managed_send_buffs_t &buffs + vrt_packet_handler::managed_send_buffs_t &buffs, + double timeout ){ UHD_ASSERT_THROW(trans.size() == buffs.size()); + bool good = true; for (size_t i = 0; i < buffs.size(); i++){ - buffs[i] = trans[i]->get_send_buff(); + buffs[i] = trans[i]->get_send_buff(timeout); + good = good and (buffs[i].get() != NULL); } - return true; + return good; } size_t usrp2_impl::send( const std::vector &buffs, size_t num_samps, const tx_metadata_t &metadata, const io_type_t &io_type, - send_mode_t send_mode + send_mode_t send_mode, double timeout ){ return vrt_packet_handler::send( _io_impl->packet_handler_send_state, //last state of the send handler @@ -213,7 +213,7 @@ size_t usrp2_impl::send( io_type, _io_helper.get_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(&get_send_buffs, _data_transports, _1), + boost::bind(&get_send_buffs, _data_transports, _1, timeout), get_max_send_samps_per_packet() ); } @@ -224,7 +224,7 @@ size_t usrp2_impl::send( size_t usrp2_impl::recv( const std::vector &buffs, size_t num_samps, rx_metadata_t &metadata, const io_type_t &io_type, - recv_mode_t recv_mode, size_t timeout_ms + recv_mode_t recv_mode, double timeout ){ return vrt_packet_handler::recv( _io_impl->packet_handler_recv_state, //last state of the recv handler @@ -233,6 +233,6 @@ size_t usrp2_impl::recv( io_type, _io_helper.get_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(), _1, timeout_ms) + boost::bind(&usrp2_impl::io_impl::get_recv_buffs, _io_impl.get(), _1, timeout) ); } diff --git a/host/lib/usrp/usrp2/usrp2_impl.hpp b/host/lib/usrp/usrp2/usrp2_impl.hpp index 157d17057..e8763b284 100644 --- a/host/lib/usrp/usrp2/usrp2_impl.hpp +++ b/host/lib/usrp/usrp2/usrp2_impl.hpp @@ -234,7 +234,7 @@ public: size_t send( const std::vector &, size_t, const uhd::tx_metadata_t &, const uhd::io_type_t &, - uhd::device::send_mode_t + uhd::device::send_mode_t, double ); size_t get_max_recv_samps_per_packet(void) const{ return _io_helper.get_max_recv_samps_per_packet(); @@ -242,9 +242,9 @@ public: size_t recv( const std::vector &, size_t, uhd::rx_metadata_t &, const uhd::io_type_t &, - uhd::device::recv_mode_t, size_t + uhd::device::recv_mode_t, double ); - bool recv_async_msg(uhd::async_metadata_t &, size_t); + bool recv_async_msg(uhd::async_metadata_t &, double); private: //device properties interface diff --git a/host/test/buffer_test.cpp b/host/test/buffer_test.cpp index aadb3f951..8445412e7 100644 --- a/host/test/buffer_test.cpp +++ b/host/test/buffer_test.cpp @@ -23,7 +23,7 @@ using namespace boost::assign; using namespace uhd::transport; -static const boost::posix_time::milliseconds timeout(10); +static const double timeout = 0.01/*secs*/; BOOST_AUTO_TEST_CASE(test_bounded_buffer_with_timed_wait){ bounded_buffer::sptr bb(bounded_buffer::make(3)); -- cgit v1.2.3 From 7352c95037fa57d37dd7adc4c2ea6935006b56c8 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sat, 2 Oct 2010 23:15:46 -0700 Subject: uhd: reworked the zero copy interface - recv buffers have a release call - safe make function for recv buffers (buff + release callback) - send buffer commits now have a void return - safe make function for send buffers (buff + commit callback) The reason for the void return from commit is that ssize_t num_bytes was never returning anything of use. That is for all of the zero copy implementations so far, commit cannot really error (being asynchronous). libusb zero copy impl was reworked to support the new interface. USRP1 io_impl with the psuedo managed buffer was replaced with safe_make. Also, usrp1 io_impl was simplified greatly due to commit returning void now. UDP zero copy asio was disabled (in this commit, until its reworked). Phony send and recv interfaces were removed completely. --- host/include/uhd/transport/zero_copy.hpp | 112 +++++++--------------- host/lib/transport/CMakeLists.txt | 4 +- host/lib/transport/libusb1_zero_copy.cpp | 149 +++++++----------------------- host/lib/transport/vrt_packet_handler.hpp | 4 +- host/lib/transport/zero_copy.cpp | 121 +++++++++--------------- host/lib/usrp/usrp1/io_impl.cpp | 65 ++++--------- 6 files changed, 130 insertions(+), 325 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/include/uhd/transport/zero_copy.hpp b/host/include/uhd/transport/zero_copy.hpp index ba19b193c..9dd16280c 100644 --- a/host/include/uhd/transport/zero_copy.hpp +++ b/host/include/uhd/transport/zero_copy.hpp @@ -19,10 +19,10 @@ #define INCLUDED_UHD_TRANSPORT_ZERO_COPY_HPP #include -#include #include #include #include +#include namespace uhd{ namespace transport{ @@ -34,14 +34,27 @@ namespace uhd{ namespace transport{ class UHD_API managed_recv_buffer : boost::noncopyable{ public: typedef boost::shared_ptr sptr; + typedef boost::function release_fcn_t; + + /*! + * Make a safe managed receive buffer: + * A safe managed buffer ensures that release is called once, + * either by the user or automatically upon deconstruction. + * \param buff a reference to the constant buffer + * \param release_fcn callback to release the memory + * \return a new managed receive buffer + */ + static sptr make_safe( + const boost::asio::const_buffer &buff, + const release_fcn_t &release_fcn + ); /*! - * 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. + * This should be called to release the buffer to the transport object. * After calling, the referenced memory should be considered invalid. */ - virtual ~managed_recv_buffer(void) = 0; + virtual void release(void) = 0; /*! * Get the size of the underlying buffer. @@ -71,20 +84,34 @@ namespace uhd{ namespace transport{ /*! * A managed send buffer: * Contains a reference to transport-managed memory, - * and a method to release the memory after writing. + * and a method to commit the memory after writing. */ class UHD_API managed_send_buffer : boost::noncopyable{ public: typedef boost::shared_ptr sptr; + typedef boost::function commit_fcn_t; + + /*! + * Make a safe managed send buffer: + * A safe managed buffer ensures that commit is called once, + * either by the user or automatically upon deconstruction. + * In the later case, the deconstructor will call commit(0). + * \param buff a reference to the mutable buffer + * \param commit_fcn callback to commit the memory + * \return a new managed send buffer + */ + static sptr make_safe( + const boost::asio::mutable_buffer &buff, + const commit_fcn_t &commit_fcn + ); /*! * 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 - * \return the number of bytes written, 0 for timeout, negative for error */ - virtual ssize_t commit(size_t num_bytes) = 0; + virtual void commit(size_t num_bytes) = 0; /*! * Get the size of the underlying buffer. @@ -154,77 +181,6 @@ namespace uhd{ namespace transport{ }; - /*! - * 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. - */ - class UHD_API phony_zero_copy_recv_if : public virtual zero_copy_if{ - public: - /*! - * Create a phony zero copy recv interface. - * \param max_buff_size max buffer size in bytes - */ - phony_zero_copy_recv_if(size_t max_buff_size); - - //! destructor - virtual ~phony_zero_copy_recv_if(void); - - /*! - * Get a new receive buffer from this transport object. - * \param timeout the timeout to get the buffer in seconds - * \return a managed buffer, or null sptr on timeout/error - */ - managed_recv_buffer::sptr get_recv_buff(double timeout); - - private: - /*! - * Perform a private copying recv. - * \param buff the buffer to write data into - * \param timeout the timeout to get the buffer in seconds - * \return the number of bytes written to buff, 0 for timeout, negative for error - */ - virtual ssize_t recv(const boost::asio::mutable_buffer &buff, double timeout) = 0; - - UHD_PIMPL_DECL(impl) _impl; - }; - - /*! - * 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. - */ - class UHD_API phony_zero_copy_send_if : public virtual zero_copy_if{ - public: - /*! - * Create a phony zero copy send interface. - * \param max_buff_size max buffer size in bytes - */ - phony_zero_copy_send_if(size_t max_buff_size); - - //! destructor - virtual ~phony_zero_copy_send_if(void); - - /*! - * Get a new send buffer from this transport object. - * \param timeout the timeout to get the buffer in seconds - * \return a managed buffer, or null sptr on timeout/error - */ - managed_send_buffer::sptr get_send_buff(double timeout); - - private: - /*! - * Perform a private copying send. - * \param buff the buffer to read data from - * \return the number of bytes read from buff, 0 for timeout, negative for error - */ - virtual ssize_t send(const boost::asio::const_buffer &buff) = 0; - - UHD_PIMPL_DECL(impl) _impl; - }; - }} //namespace #endif /* INCLUDED_UHD_TRANSPORT_ZERO_COPY_HPP */ diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt index 61616d077..2be2c89ec 100644 --- a/host/lib/transport/CMakeLists.txt +++ b/host/lib/transport/CMakeLists.txt @@ -36,6 +36,8 @@ IF(LIBUSB_FOUND) INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/lib/transport/msvc) ENDIF(MSVC) SET(HAVE_USB_SUPPORT TRUE) +ELSE(LIBUSB_FOUND) + #TODO dummy usb ENDIF(LIBUSB_FOUND) IF(HAVE_USB_SUPPORT) @@ -99,7 +101,7 @@ SET_SOURCE_FILES_PROPERTIES( LIBUHD_APPEND_SOURCES( ${CMAKE_SOURCE_DIR}/lib/transport/if_addrs.cpp ${CMAKE_SOURCE_DIR}/lib/transport/udp_simple.cpp - ${CMAKE_SOURCE_DIR}/lib/transport/udp_zero_copy_asio.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/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index 7f2bc3468..e1cc8398c 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -95,9 +95,6 @@ private: int _endpoint; bool _input; - size_t _transfer_size; - size_t _num_transfers; - //! hold a bounded buffer of completed transfers typedef bounded_buffer lut_buff_type; lut_buff_type::sptr _completed_list; @@ -154,14 +151,12 @@ usb_endpoint::usb_endpoint( ): _handle(handle), _endpoint(endpoint), - _input(input), - _transfer_size(transfer_size), - _num_transfers(num_transfers) + _input(input) { _completed_list = lut_buff_type::make(num_transfers); - for (size_t i = 0; i < _num_transfers; i++){ - _all_luts.push_back(allocate_transfer(_transfer_size)); + for (size_t i = 0; i < num_transfers; i++){ + _all_luts.push_back(allocate_transfer(transfer_size)); //input luts are immediately submitted to be filled //output luts go into the completed list as free buffers @@ -280,126 +275,44 @@ libusb_transfer *usb_endpoint::get_lut_with_wait(double timeout){ } /*********************************************************************** - * Managed buffers + * USB zero_copy device class **********************************************************************/ -/* - * Libusb managed receive buffer - * Construct a recv buffer from a libusb transfer. The memory held by - * the libusb transfer is exposed through the managed buffer interface. - * Upon destruction, the transfer and buffer are resubmitted to the - * endpoint for further use. - */ -class libusb_managed_recv_buffer_impl : public managed_recv_buffer { +class libusb_zero_copy_impl : public usb_zero_copy { public: - libusb_managed_recv_buffer_impl(libusb_transfer *lut, - usb_endpoint::sptr endpoint) - : _buff(lut->buffer, lut->length) - { - _lut = lut; - _endpoint = endpoint; - } - - ~libusb_managed_recv_buffer_impl(void){ - _endpoint->submit(_lut); - } + typedef boost::shared_ptr sptr; -private: - const boost::asio::const_buffer &get(void) const{ - return _buff; - } + libusb_zero_copy_impl( + libusb::device_handle::sptr handle, + unsigned int recv_endpoint, unsigned int send_endpoint, + size_t recv_xfer_size, size_t recv_num_xfers, + size_t send_xfer_size, size_t send_num_xfers + ); - libusb_transfer *_lut; - usb_endpoint::sptr _endpoint; - const boost::asio::const_buffer _buff; -}; + managed_recv_buffer::sptr get_recv_buff(double); + managed_send_buffer::sptr get_send_buff(double); -/* - * Libusb managed send buffer - * Construct a send buffer from a libusb transfer. The memory held by - * the libusb transfer is exposed through the managed buffer interface. - * Committing the buffer will set the data length and submit the buffer - * to the endpoint. Submitting a buffer multiple times or destroying - * the buffer before committing is an error. For the latter, the transfer - * is returned to the endpoint with no data for reuse. - */ -class libusb_managed_send_buffer_impl : public managed_send_buffer { -public: - libusb_managed_send_buffer_impl(libusb_transfer *lut, - usb_endpoint::sptr endpoint) - : _buff(lut->buffer, lut->length), _committed(false) - { - _lut = lut; - _endpoint = endpoint; - } + size_t get_num_recv_frames(void) const { return _recv_num_frames; } + size_t get_num_send_frames(void) const { return _send_num_frames; } - ~libusb_managed_send_buffer_impl(void){ - if (!_committed) { - _lut->length = 0; - _lut->actual_length = 0; - _endpoint->submit(_lut); - } +private: + void release(libusb_transfer *lut){ + _recv_ep->submit(lut); } - ssize_t commit(size_t num_bytes) - { - if (_committed) { - std::cerr << "UHD: send buffer already committed" << std::endl; - return 0; - } - - UHD_ASSERT_THROW(num_bytes <= boost::asio::buffer_size(_buff)); - - _lut->length = num_bytes; - _lut->actual_length = 0; - + void commit(libusb_transfer *lut, size_t num_bytes){ + lut->length = num_bytes; try{ - _endpoint->submit(_lut); - _committed = true; - return num_bytes; + _send_ep->submit(lut); } catch(const std::exception &e){ std::cerr << "Error in commit: " << e.what() << std::endl; - return -1; } } -private: - const boost::asio::mutable_buffer &get(void) const{ - return _buff; - } - - libusb_transfer *_lut; - usb_endpoint::sptr _endpoint; - const boost::asio::mutable_buffer _buff; - bool _committed; -}; - - -/*********************************************************************** - * USB zero_copy device class - **********************************************************************/ -class libusb_zero_copy_impl : public usb_zero_copy -{ -private: libusb::device_handle::sptr _handle; + size_t _recv_xfer_size, _send_xfer_size; size_t _recv_num_frames, _send_num_frames; usb_endpoint::sptr _recv_ep, _send_ep; - -public: - typedef boost::shared_ptr sptr; - - libusb_zero_copy_impl( - libusb::device_handle::sptr handle, - unsigned int recv_endpoint, unsigned int send_endpoint, - size_t recv_xfer_size, size_t recv_num_xfers, - size_t send_xfer_size, size_t send_num_xfers - ); - - managed_recv_buffer::sptr get_recv_buff(double); - managed_send_buffer::sptr get_send_buff(double); - - size_t get_num_recv_frames(void) const { return _recv_num_frames; } - size_t get_num_send_frames(void) const { return _send_num_frames; } }; /* @@ -426,7 +339,9 @@ libusb_zero_copy_impl::libusb_zero_copy_impl( UHD_ASSERT_THROW(send_xfer_size % 512 == 0); //store the num xfers for the num frames count + _recv_xfer_size = recv_xfer_size; _recv_num_frames = recv_num_xfers; + _send_xfer_size = send_xfer_size; _send_num_frames = send_num_xfers; _handle->claim_interface(2 /*in interface*/); @@ -461,9 +376,10 @@ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(double timeout){ return managed_recv_buffer::sptr(); } else { - return managed_recv_buffer::sptr( - new libusb_managed_recv_buffer_impl(lut, - _recv_ep)); + return managed_recv_buffer::make_safe( + boost::asio::const_buffer(lut->buffer, lut->actual_length), + boost::bind(&libusb_zero_copy_impl::release, this, lut) + ); } } @@ -480,9 +396,10 @@ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){ return managed_send_buffer::sptr(); } else { - return managed_send_buffer::sptr( - new libusb_managed_send_buffer_impl(lut, - _send_ep)); + return managed_send_buffer::make_safe( + boost::asio::mutable_buffer(lut->buffer, _send_xfer_size), + boost::bind(&libusb_zero_copy_impl::commit, this, lut, _1) + ); } } diff --git a/host/lib/transport/vrt_packet_handler.hpp b/host/lib/transport/vrt_packet_handler.hpp index e11afff30..939517411 100644 --- a/host/lib/transport/vrt_packet_handler.hpp +++ b/host/lib/transport/vrt_packet_handler.hpp @@ -343,9 +343,7 @@ template UHD_INLINE T get_context_code( //commit the samples to the zero-copy interface size_t num_bytes_total = (vrt_header_offset_words32+if_packet_info.num_packet_words32)*sizeof(boost::uint32_t); - if (send_buffs[i]->commit(num_bytes_total) < ssize_t(num_bytes_total)){ - std::cerr << "commit to send buffer returned less than commit size" << std::endl; - } + send_buffs[i]->commit(num_bytes_total); } return num_samps; } diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp index dfb65951f..a5a864a04 100644 --- a/host/lib/transport/zero_copy.cpp +++ b/host/lib/transport/zero_copy.cpp @@ -16,32 +16,35 @@ // #include -#include -#include -#include using namespace uhd::transport; /*********************************************************************** - * The pure-virtual deconstructor needs an implementation to be happy + * Safe managed receive buffer **********************************************************************/ -managed_recv_buffer::~managed_recv_buffer(void){ +static void release_nop(void){ /* NOP */ } -/*********************************************************************** - * Phony zero-copy recv interface implementation - **********************************************************************/ - -//! phony zero-copy recv buffer implementation -class managed_recv_buffer_impl : public managed_recv_buffer{ +class safe_managed_receive_buffer : public managed_recv_buffer{ public: - managed_recv_buffer_impl(const boost::asio::const_buffer &buff) : _buff(buff){ + safe_managed_receive_buffer( + const boost::asio::const_buffer &buff, + const release_fcn_t &release_fcn + ): + _buff(buff), _release_fcn(release_fcn) + { /* NOP */ } - ~managed_recv_buffer_impl(void){ - delete [] this->cast(); + ~safe_managed_receive_buffer(void){ + _release_fcn(); + } + + void release(void){ + release_fcn_t release_fcn = _release_fcn; + _release_fcn = &release_nop; + return release_fcn(); } private: @@ -50,64 +53,42 @@ private: } const boost::asio::const_buffer _buff; + release_fcn_t _release_fcn; }; -//! phony zero-copy recv interface implementation -struct phony_zero_copy_recv_if::impl{ - impl(size_t max_buff_size) : max_buff_size(max_buff_size){ - /* NOP */ - } - size_t max_buff_size; -}; - -phony_zero_copy_recv_if::phony_zero_copy_recv_if(size_t max_buff_size){ - _impl = UHD_PIMPL_MAKE(impl, (max_buff_size)); -} - -phony_zero_copy_recv_if::~phony_zero_copy_recv_if(void){ - /* NOP */ -} - -managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(double timeout){ - //allocate memory - boost::uint8_t *recv_mem = new boost::uint8_t[_impl->max_buff_size]; - - //call recv() with timeout option - ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout); - - if (num_bytes <= 0) return managed_recv_buffer::sptr(); //NULL sptr - - //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_recv_buffer::sptr managed_recv_buffer::make_safe( + const boost::asio::const_buffer &buff, + const release_fcn_t &release_fcn +){ + return sptr(new safe_managed_receive_buffer(buff, release_fcn)); } /*********************************************************************** - * Phony zero-copy send interface implementation + * Safe managed send buffer **********************************************************************/ +static void commit_nop(size_t){ + /* NOP */ +} -//! phony zero-copy send buffer implementation -class managed_send_buffer_impl : public managed_send_buffer{ +class safe_managed_send_buffer : public managed_send_buffer{ public: - typedef boost::function send_fcn_t; - - managed_send_buffer_impl( + safe_managed_send_buffer( const boost::asio::mutable_buffer &buff, - const send_fcn_t &send_fcn + const commit_fcn_t &commit_fcn ): - _buff(buff), - _send_fcn(send_fcn) + _buff(buff), _commit_fcn(commit_fcn) { /* NOP */ } - ~managed_send_buffer_impl(void){ - /* NOP */ + ~safe_managed_send_buffer(void){ + _commit_fcn(0); } - ssize_t commit(size_t num_bytes){ - return _send_fcn(boost::asio::buffer(_buff, num_bytes)); + void commit(size_t num_bytes){ + commit_fcn_t commit_fcn = _commit_fcn; + _commit_fcn = &commit_nop; + return commit_fcn(num_bytes); } private: @@ -116,28 +97,12 @@ private: } 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; + commit_fcn_t _commit_fcn; }; -phony_zero_copy_send_if::phony_zero_copy_send_if(size_t max_buff_size){ - _impl = UHD_PIMPL_MAKE(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; -} - -managed_send_buffer::sptr phony_zero_copy_send_if::get_send_buff(double){ - return _impl->send_buff; //FIXME there is only ever one send buff, we assume that the caller doesnt hang onto these +safe_managed_send_buffer::sptr managed_send_buffer::make_safe( + const boost::asio::mutable_buffer &buff, + const commit_fcn_t &commit_fcn +){ + return sptr(new safe_managed_send_buffer(buff, commit_fcn)); } diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp index 8d9c68961..676b1536a 100644 --- a/host/lib/usrp/usrp1/io_impl.cpp +++ b/host/lib/usrp/usrp1/io_impl.cpp @@ -33,35 +33,6 @@ using namespace uhd::usrp; using namespace uhd::transport; namespace asio = boost::asio; -/*********************************************************************** - * Pseudo send buffer implementation - **********************************************************************/ -class pseudo_managed_send_buffer : public managed_send_buffer{ -public: - - pseudo_managed_send_buffer( - const boost::asio::mutable_buffer &buff, - const boost::function &commit - ): - _buff(buff), - _commit(commit) - { - /* NOP */ - } - - ssize_t commit(size_t num_bytes){ - return _commit(num_bytes); - } - -private: - const boost::asio::mutable_buffer &get(void) const{ - return _buff; - } - - const boost::asio::mutable_buffer _buff; - const boost::function _commit; -}; - /*********************************************************************** * IO Implementation Details **********************************************************************/ @@ -96,9 +67,9 @@ struct usrp1_impl::io_impl{ size_t num_bytes_committed; double send_timeout; boost::uint8_t pseudo_buff[BYTES_PER_PACKET]; - ssize_t phony_commit_pseudo_buff(size_t num_bytes); - ssize_t phony_commit_send_buff(size_t num_bytes); - ssize_t commit_send_buff(void); + void phony_commit_pseudo_buff(size_t num_bytes); + void phony_commit_send_buff(size_t num_bytes); + void commit_send_buff(void); void flush_send_buff(void); bool get_send_buffs(vrt_packet_handler::managed_send_buffs_t &, double); @@ -119,28 +90,25 @@ struct usrp1_impl::io_impl{ * The first loop iteration will fill the remainder of the send buffer. * The second loop iteration will empty the pseudo buffer remainder. */ -ssize_t usrp1_impl::io_impl::phony_commit_pseudo_buff(size_t num_bytes){ +void usrp1_impl::io_impl::phony_commit_pseudo_buff(size_t num_bytes){ size_t bytes_to_copy = num_bytes, bytes_copied = 0; while(bytes_to_copy){ size_t bytes_copied_here = std::min(bytes_to_copy, get_send_mem_size()); std::memcpy(get_send_mem_ptr(), pseudo_buff + bytes_copied, bytes_copied_here); - ssize_t ret = phony_commit_send_buff(bytes_copied_here); - if (ret < 0) return ret; - bytes_to_copy -= ret; - bytes_copied += ret; + phony_commit_send_buff(bytes_copied_here); + bytes_to_copy -= bytes_copied_here; + bytes_copied += bytes_copied_here; } - return bytes_copied; } /*! * Accept a commit of num bytes to the send buffer. * Conditionally commit the send buffer if full. */ -ssize_t usrp1_impl::io_impl::phony_commit_send_buff(size_t num_bytes){ +void usrp1_impl::io_impl::phony_commit_send_buff(size_t num_bytes){ num_bytes_committed += num_bytes; - if (num_bytes_committed != send_buff->size()) return num_bytes; - ssize_t ret = commit_send_buff(); - return (ret < 0)? ret : num_bytes; + if (num_bytes_committed != send_buff->size()) return; + commit_send_buff(); } /*! @@ -158,11 +126,10 @@ void usrp1_impl::io_impl::flush_send_buff(void){ * Perform an actual commit on the send buffer: * Commit the contents of the send buffer and request a new buffer. */ -ssize_t usrp1_impl::io_impl::commit_send_buff(void){ - ssize_t ret = send_buff->commit(num_bytes_committed); +void usrp1_impl::io_impl::commit_send_buff(void){ + send_buff->commit(num_bytes_committed); send_buff = data_transport->get_send_buff(send_timeout); num_bytes_committed = 0; - return ret; } bool usrp1_impl::io_impl::get_send_buffs( @@ -173,17 +140,17 @@ bool usrp1_impl::io_impl::get_send_buffs( //not enough bytes free -> use the pseudo buffer if (get_send_mem_size() < BYTES_PER_PACKET){ - buffs[0] = managed_send_buffer::sptr(new pseudo_managed_send_buffer( + buffs[0] = managed_send_buffer::make_safe( boost::asio::buffer(pseudo_buff), boost::bind(&usrp1_impl::io_impl::phony_commit_pseudo_buff, this, _1) - )); + ); } //otherwise use the send buffer offset by the bytes written else{ - buffs[0] = managed_send_buffer::sptr(new pseudo_managed_send_buffer( + buffs[0] = managed_send_buffer::make_safe( boost::asio::buffer(get_send_mem_ptr(), get_send_mem_size()), boost::bind(&usrp1_impl::io_impl::phony_commit_send_buff, this, _1) - )); + ); } return buffs[0].get() != NULL; -- cgit v1.2.3 From b57c84b34bcdd6c66eb053695b83e6bd6c481774 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sun, 3 Oct 2010 01:41:24 -0700 Subject: uhd: implemented udp zero copy asio with async calls --- host/lib/transport/CMakeLists.txt | 2 +- host/lib/transport/libusb1_zero_copy.cpp | 6 +- host/lib/transport/udp_zero_copy_asio.cpp | 186 +++++++++++++++++++++--------- 3 files changed, 133 insertions(+), 61 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt index 2be2c89ec..bf92b0822 100644 --- a/host/lib/transport/CMakeLists.txt +++ b/host/lib/transport/CMakeLists.txt @@ -101,7 +101,7 @@ SET_SOURCE_FILES_PROPERTIES( LIBUHD_APPEND_SOURCES( ${CMAKE_SOURCE_DIR}/lib/transport/if_addrs.cpp ${CMAKE_SOURCE_DIR}/lib/transport/udp_simple.cpp - #${CMAKE_SOURCE_DIR}/lib/transport/udp_zero_copy_asio.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/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index e1cc8398c..c819302b6 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -277,7 +277,7 @@ libusb_transfer *usb_endpoint::get_lut_with_wait(double timeout){ /*********************************************************************** * USB zero_copy device class **********************************************************************/ -class libusb_zero_copy_impl : public usb_zero_copy { +class libusb_zero_copy_impl : public usb_zero_copy, public boost::enable_shared_from_this { public: typedef boost::shared_ptr sptr; @@ -378,7 +378,7 @@ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(double timeout){ else { return managed_recv_buffer::make_safe( boost::asio::const_buffer(lut->buffer, lut->actual_length), - boost::bind(&libusb_zero_copy_impl::release, this, lut) + boost::bind(&libusb_zero_copy_impl::release, shared_from_this(), lut) ); } } @@ -398,7 +398,7 @@ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){ else { return managed_send_buffer::make_safe( boost::asio::mutable_buffer(lut->buffer, _send_xfer_size), - boost::bind(&libusb_zero_copy_impl::commit, this, lut, _1) + boost::bind(&libusb_zero_copy_impl::commit, shared_from_this(), lut, _1) ); } } diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 3130830a5..70e7514a1 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -17,20 +17,23 @@ #include #include //mtu +#include #include #include -#include +#include #include #include +#include #include using namespace uhd::transport; +namespace asio = boost::asio; /*********************************************************************** * Constants **********************************************************************/ //enough buffering for half a second of samples at full rate on usrp2 -static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(sizeof(boost::uint32_t) * 25e6 * 0.5); +static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(4 * 25e6 * 0.5); //Large buffers cause more underflow at high rates. //Perhaps this is due to the kernel scheduling, //but may change with host-based flow control. @@ -43,39 +46,55 @@ static const size_t MIN_SEND_SOCK_BUFF_SIZE = size_t(10e3); * 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 phony_zero_copy_recv_if, - public phony_zero_copy_send_if, - public udp_zero_copy -{ +class udp_zero_copy_asio_impl : public udp_zero_copy, public boost::enable_shared_from_this { public: - typedef boost::shared_ptr sptr; - - udp_zero_copy_impl( - const std::string &addr, - const std::string &port - ): - phony_zero_copy_recv_if(udp_simple::mtu), - phony_zero_copy_send_if(udp_simple::mtu) - { + typedef boost::shared_ptr sptr; + + udp_zero_copy_asio_impl(const std::string &addr, const std::string &port){ //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl; // resolve the address - boost::asio::ip::udp::resolver resolver(_io_service); - boost::asio::ip::udp::resolver::query query(boost::asio::ip::udp::v4(), addr, port); - boost::asio::ip::udp::endpoint receiver_endpoint = *resolver.resolve(query); + asio::ip::udp::resolver resolver(_io_service); + asio::ip::udp::resolver::query query(asio::ip::udp::v4(), addr, port); + asio::ip::udp::endpoint receiver_endpoint = *resolver.resolve(query); // create, open, and connect the socket - _socket = new boost::asio::ip::udp::socket(_io_service); - _socket->open(boost::asio::ip::udp::v4()); + _socket = new asio::ip::udp::socket(_io_service); + _socket->open(asio::ip::udp::v4()); _socket->connect(receiver_endpoint); - _sock_fd = _socket->native(); } - ~udp_zero_copy_impl(void){ + ~udp_zero_copy_asio_impl(void){ + _io_service.stop(); + _thread_group.join_all(); delete _socket; } + /*! + * Init, the second contructor: + * Allocate memory and spwan service thread. + */ + void init(void){ + //allocate all recv frames and release them to begin xfers + _pending_recv_buffs = pending_buffs_type::make(this->get_num_recv_frames()); + for (size_t i = 0; i < this->get_num_recv_frames(); i++){ + boost::shared_array buff(new char[udp_simple::mtu]); + _buffers.push_back(buff); //store a reference to this shared array + release(buff.get()); + } + + //allocate all send frames and push them into the fifo + _pending_send_buffs = pending_buffs_type::make(this->get_num_send_frames()); + for (size_t i = 0; i < this->get_num_send_frames(); i++){ + boost::shared_array buff(new char[udp_simple::mtu]); + _buffers.push_back(buff); //store a reference to this shared array + handle_send(buff.get()); + } + + //spawn the service thread that will run the io service + _thread_group.create_thread(boost::bind(&udp_zero_copy_asio_impl::service, this)); + } + //get size for internal socket buffer template size_t get_buff_size(void) const{ Opt option; @@ -90,60 +109,111 @@ public: return get_buff_size(); } - //The number of frames is approximately the buffer size divided by the max datagram size. //In reality, this is a phony zero-copy interface and the number of frames is infinite. //However, its sensible to advertise a frame count that is approximate to buffer size. //This way, the transport caller will have an idea about how much buffering to create. size_t get_num_recv_frames(void) const{ - return this->get_buff_size()/udp_simple::mtu; + return this->get_buff_size()/udp_simple::mtu; } size_t get_num_send_frames(void) const{ - return this->get_buff_size()/udp_simple::mtu; + return this->get_buff_size()/udp_simple::mtu; + } + + //! pop a filled recv buffer off of the fifo and bind with the release callback + 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; + if (_pending_recv_buffs->pop_with_timed_wait(buff, timeout)){ + return managed_recv_buffer::make_safe( + buff, boost::bind( + &udp_zero_copy_asio_impl::release, + shared_from_this(), + asio::buffer_cast(buff) + ) + ); + } + return managed_recv_buffer::sptr(); + } + + //! 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 + asio::mutable_buffer buff; + if (_pending_send_buffs->pop_with_timed_wait(buff, timeout)){ + return managed_send_buffer::make_safe( + buff, boost::bind( + &udp_zero_copy_asio_impl::commit, + shared_from_this(), + asio::buffer_cast(buff), _1 + ) + ); + } + return managed_send_buffer::sptr(); } private: - boost::asio::ip::udp::socket *_socket; - boost::asio::io_service _io_service; - int _sock_fd; - - ssize_t recv(const boost::asio::mutable_buffer &buff, double timeout){ - //setup timeval for timeout - timeval tv; - tv.tv_sec = 0; - tv.tv_usec = long(timeout*1e6); - - //setup rset for timeout - fd_set rset; - 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 0; - - return ::recv( - _sock_fd, - boost::asio::buffer_cast(buff), - boost::asio::buffer_size(buff), 0 + void service(void){ + _io_service.run(); + } + + /******************************************************************* + * The async send and receive callbacks + ******************************************************************/ + + //! 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, udp_simple::mtu), + boost::bind( + &udp_zero_copy_asio_impl::handle_recv, + shared_from_this(), mem, + asio::placeholders::bytes_transferred + ) ); } - ssize_t send(const boost::asio::const_buffer &buff){ - return ::send( - _sock_fd, - boost::asio::buffer_cast(buff), - boost::asio::buffer_size(buff), 0 + //! 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, udp_simple::mtu)); + } + + //! commit a send buffer -> start an async send on the buffer + void commit(void *mem, size_t len){ + _socket->async_send( + boost::asio::buffer(mem, len), + boost::bind( + &udp_zero_copy_asio_impl::handle_send, + shared_from_this(), mem + ) ); } + + //asio guts -> socket and service + asio::ip::udp::socket *_socket; + asio::io_service _io_service; + + //memory management -> buffers and fifos + boost::thread_group _thread_group; + std::vector > _buffers; + typedef bounded_buffer pending_buffs_type; + pending_buffs_type::sptr _pending_recv_buffs, _pending_send_buffs; }; /*********************************************************************** * UDP zero copy make function **********************************************************************/ template static void resize_buff_helper( - udp_zero_copy_impl::sptr udp_trans, + udp_zero_copy_asio_impl::sptr udp_trans, size_t target_size, const std::string &name ){ @@ -183,11 +253,13 @@ udp_zero_copy::sptr udp_zero_copy::make( size_t recv_buff_size, size_t send_buff_size ){ - udp_zero_copy_impl::sptr udp_trans(new udp_zero_copy_impl(addr, port)); + udp_zero_copy_asio_impl::sptr udp_trans(new udp_zero_copy_asio_impl(addr, port)); //call the helper to resize send and recv buffers - resize_buff_helper(udp_trans, recv_buff_size, "recv"); - resize_buff_helper (udp_trans, send_buff_size, "send"); + resize_buff_helper(udp_trans, recv_buff_size, "recv"); + resize_buff_helper (udp_trans, send_buff_size, "send"); + + udp_trans->init(); //buffers resized -> call init() to use return udp_trans; } -- cgit v1.2.3 From c6f17f1e2f908747ae1547fa43b2c22c3c20ba50 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sun, 3 Oct 2010 19:34:41 -0700 Subject: uhd: changed buffer allocations to be in a single chunk, udp: pass frame sizes into the impl constructor --- host/lib/transport/libusb1_zero_copy.cpp | 27 +++++----- host/lib/transport/udp_zero_copy_asio.cpp | 85 ++++++++++++++++--------------- 2 files changed, 59 insertions(+), 53 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index c819302b6..ab48e4fc4 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -100,13 +100,13 @@ private: lut_buff_type::sptr _completed_list; //! a list of all transfer structs we allocated - std::vector _all_luts; + std::vector _all_luts; - //! a list of shared arrays for the transfer buffers - std::vector > _buffers; + //! a block of memory for the transfer buffers + boost::shared_array _buffer; // Calls for processing asynchronous I/O - libusb_transfer *allocate_transfer(int buff_len); + libusb_transfer *allocate_transfer(void *mem, size_t len); void print_transfer_status(libusb_transfer *lut); }; @@ -154,9 +154,9 @@ usb_endpoint::usb_endpoint( _input(input) { _completed_list = lut_buff_type::make(num_transfers); - + _buffer = boost::shared_array(new char[num_transfers*transfer_size]); for (size_t i = 0; i < num_transfers; i++){ - _all_luts.push_back(allocate_transfer(transfer_size)); + _all_luts.push_back(allocate_transfer(_buffer.get() + i*transfer_size, transfer_size)); //input luts are immediately submitted to be filled //output luts go into the completed list as free buffers @@ -193,23 +193,23 @@ usb_endpoint::~usb_endpoint(void){ * Allocate a libusb transfer * The allocated transfer - and buffer it contains - is repeatedly * submitted, reaped, and reused and should not be freed until shutdown. - * \param buff_len size of the individual buffer held by each transfer + * \param mem a pointer to the buffer memory + * \param len size of the individual buffer * \return pointer to an allocated libusb_transfer */ -libusb_transfer *usb_endpoint::allocate_transfer(int buff_len){ +libusb_transfer *usb_endpoint::allocate_transfer(void *mem, size_t len){ libusb_transfer *lut = libusb_alloc_transfer(0); - - boost::shared_array buff(new boost::uint8_t[buff_len]); - _buffers.push_back(buff); //store a reference to this shared array + UHD_ASSERT_THROW(lut != NULL); unsigned int endpoint = ((_endpoint & 0x7f) | (_input ? 0x80 : 0)); + unsigned char *buff = reinterpret_cast(mem); libusb_transfer_cb_fn lut_callback = libusb_transfer_cb_fn(&callback); libusb_fill_bulk_transfer(lut, // transfer _handle->get(), // dev_handle endpoint, // endpoint - buff.get(), // buffer - buff_len, // length + buff, // buffer + len, // length lut_callback, // callback this, // user_data 0); // timeout @@ -232,6 +232,7 @@ void usb_endpoint::submit(libusb_transfer *lut){ * \param lut pointer to an libusb_transfer */ void usb_endpoint::print_transfer_status(libusb_transfer *lut){ + std::cout << "here " << lut->status << std::endl; switch (lut->status) { case LIBUSB_TRANSFER_COMPLETED: if (lut->actual_length < lut->length) { diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 70e7514a1..a1eb516fc 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -18,6 +18,7 @@ #include #include //mtu #include +#include #include #include #include @@ -26,6 +27,7 @@ #include #include +using namespace uhd; using namespace uhd::transport; namespace asio = boost::asio; @@ -34,11 +36,15 @@ namespace asio = boost::asio; **********************************************************************/ //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); + //Large buffers cause more underflow at high rates. //Perhaps this is due to the kernel scheduling, //but may change with host-based flow control. static const size_t MIN_SEND_SOCK_BUFF_SIZE = size_t(10e3); +//the number of async frames to allocate for each send and recv +static const size_t DEFAULT_NUM_ASYNC_FRAMES = 32; + /*********************************************************************** * Zero Copy UDP implementation with ASIO: * This is the portable zero copy implementation for systems @@ -50,51 +56,52 @@ class udp_zero_copy_asio_impl : public udp_zero_copy, public boost::enable_share public: typedef boost::shared_ptr sptr; - udp_zero_copy_asio_impl(const std::string &addr, const std::string &port){ + udp_zero_copy_asio_impl( + const std::string &addr, const std::string &port, + size_t recv_frame_size, size_t num_recv_frames, + size_t send_frame_size, size_t num_send_frames + ): + _recv_frame_size(recv_frame_size), _num_recv_frames(num_recv_frames), + _send_frame_size(send_frame_size), _num_send_frames(num_send_frames) + { //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl; - // resolve the address + //resolve the address asio::ip::udp::resolver resolver(_io_service); asio::ip::udp::resolver::query query(asio::ip::udp::v4(), addr, port); asio::ip::udp::endpoint receiver_endpoint = *resolver.resolve(query); - // create, open, and connect the socket + //create, open, and connect the socket _socket = new asio::ip::udp::socket(_io_service); _socket->open(asio::ip::udp::v4()); _socket->connect(receiver_endpoint); } - ~udp_zero_copy_asio_impl(void){ - _io_service.stop(); - _thread_group.join_all(); - delete _socket; - } - - /*! - * Init, the second contructor: - * Allocate memory and spwan service thread. - */ void init(void){ //allocate all recv frames and release them to begin xfers - _pending_recv_buffs = pending_buffs_type::make(this->get_num_recv_frames()); - for (size_t i = 0; i < this->get_num_recv_frames(); i++){ - boost::shared_array buff(new char[udp_simple::mtu]); - _buffers.push_back(buff); //store a reference to this shared array - release(buff.get()); + _pending_recv_buffs = pending_buffs_type::make(_num_recv_frames); + _recv_buffer = boost::shared_array(new char[_num_recv_frames*_recv_frame_size]); + for (size_t i = 0; i < _num_recv_frames; i++){ + release(_recv_buffer.get() + i*_recv_frame_size); } //allocate all send frames and push them into the fifo - _pending_send_buffs = pending_buffs_type::make(this->get_num_send_frames()); - for (size_t i = 0; i < this->get_num_send_frames(); i++){ - boost::shared_array buff(new char[udp_simple::mtu]); - _buffers.push_back(buff); //store a reference to this shared array - handle_send(buff.get()); + _pending_send_buffs = pending_buffs_type::make(_num_send_frames); + _send_buffer = boost::shared_array(new char[_num_send_frames*_send_frame_size]); + for (size_t i = 0; i < _num_send_frames; i++){ + handle_send(_send_buffer.get() + i*_send_frame_size); } //spawn the service thread that will run the io service _thread_group.create_thread(boost::bind(&udp_zero_copy_asio_impl::service, this)); } + ~udp_zero_copy_asio_impl(void){ + _io_service.stop(); + _thread_group.join_all(); + delete _socket; + } + //get size for internal socket buffer template size_t get_buff_size(void) const{ Opt option; @@ -109,19 +116,6 @@ public: return get_buff_size(); } - //The number of frames is approximately the buffer size divided by the max datagram size. - //In reality, this is a phony zero-copy interface and the number of frames is infinite. - //However, its sensible to advertise a frame count that is approximate to buffer size. - //This way, the transport caller will have an idea about how much buffering to create. - - size_t get_num_recv_frames(void) const{ - return this->get_buff_size()/udp_simple::mtu; - } - - size_t get_num_send_frames(void) const{ - return this->get_buff_size()/udp_simple::mtu; - } - //! pop a filled recv buffer off of the fifo and bind with the release callback managed_recv_buffer::sptr get_recv_buff(double timeout){ boost::this_thread::disable_interruption di; //disable because the wait can throw @@ -138,6 +132,8 @@ public: return managed_recv_buffer::sptr(); } + size_t get_num_recv_frames(void) const {return _num_recv_frames;} + //! 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 @@ -154,8 +150,11 @@ public: return managed_send_buffer::sptr(); } + size_t get_num_send_frames(void) const {return _num_send_frames;} + private: void service(void){ + set_thread_priority_safe(); _io_service.run(); } @@ -172,7 +171,7 @@ private: //! release a recv buffer -> start an async recv on the buffer void release(void *mem){ _socket->async_receive( - boost::asio::buffer(mem, udp_simple::mtu), + boost::asio::buffer(mem, _recv_frame_size), boost::bind( &udp_zero_copy_asio_impl::handle_recv, shared_from_this(), mem, @@ -184,7 +183,7 @@ private: //! 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, udp_simple::mtu)); + _pending_send_buffs->push_with_wait(boost::asio::buffer(mem, _send_frame_size)); } //! commit a send buffer -> start an async send on the buffer @@ -204,9 +203,11 @@ private: //memory management -> buffers and fifos boost::thread_group _thread_group; - std::vector > _buffers; + boost::shared_array _send_buffer, _recv_buffer; typedef bounded_buffer pending_buffs_type; pending_buffs_type::sptr _pending_recv_buffs, _pending_send_buffs; + const size_t _recv_frame_size, _num_recv_frames; + const size_t _send_frame_size, _num_send_frames; }; /*********************************************************************** @@ -253,7 +254,11 @@ udp_zero_copy::sptr udp_zero_copy::make( size_t recv_buff_size, size_t send_buff_size ){ - udp_zero_copy_asio_impl::sptr udp_trans(new udp_zero_copy_asio_impl(addr, port)); + udp_zero_copy_asio_impl::sptr udp_trans(new udp_zero_copy_asio_impl( + addr, port, + udp_simple::mtu, DEFAULT_NUM_ASYNC_FRAMES, //recv + udp_simple::mtu, DEFAULT_NUM_ASYNC_FRAMES //send + )); //call the helper to resize send and recv buffers resize_buff_helper(udp_trans, recv_buff_size, "recv"); -- cgit v1.2.3 From 219183453b23a572ddd6eeec81c3bc1907754560 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Mon, 4 Oct 2010 11:30:14 -0700 Subject: uhd: added include for enable_shared_from_this when used --- host/lib/transport/libusb1_zero_copy.cpp | 2 +- host/lib/transport/udp_zero_copy_asio.cpp | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index ab48e4fc4..c7f084f62 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include @@ -280,7 +281,6 @@ libusb_transfer *usb_endpoint::get_lut_with_wait(double timeout){ **********************************************************************/ class libusb_zero_copy_impl : public usb_zero_copy, public boost::enable_shared_from_this { public: - typedef boost::shared_ptr sptr; libusb_zero_copy_impl( libusb::device_handle::sptr handle, diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 2cf7bde18..e9d91fe45 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include using namespace uhd; -- cgit v1.2.3 From 5bd863efa05f3c6cfd672cffe87f57b33d1c32b5 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Mon, 4 Oct 2010 14:54:16 -0700 Subject: uhd: bounded buffer pop sets back element to empty to decrement references --- host/include/uhd/transport/bounded_buffer.ipp | 16 ++++++++++++++-- host/lib/transport/libusb1_zero_copy.cpp | 4 ---- 2 files changed, 14 insertions(+), 6 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/include/uhd/transport/bounded_buffer.ipp b/host/include/uhd/transport/bounded_buffer.ipp index 71143741e..58f78bab4 100644 --- a/host/include/uhd/transport/bounded_buffer.ipp +++ b/host/include/uhd/transport/bounded_buffer.ipp @@ -73,7 +73,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE void pop_with_wait(elem_type &elem){ boost::unique_lock lock(_mutex); _empty_cond.wait(lock, boost::bind(&bounded_buffer_impl::not_empty, this)); - elem = _buffer.back(); _buffer.pop_back(); + this->pop_back(elem); lock.unlock(); _full_cond.notify_one(); } @@ -84,7 +84,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ lock, boost::posix_time::microseconds(long(timeout*1e6)), boost::bind(&bounded_buffer_impl::not_empty, this) )) return false; - elem = _buffer.back(); _buffer.pop_back(); + this->pop_back(elem); lock.unlock(); _full_cond.notify_one(); return true; @@ -104,6 +104,18 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ bool not_full(void) const{return not _buffer.full();} bool not_empty(void) const{return not _buffer.empty();} + + /*! + * 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(); + _buffer.back() = elem_type(); + _buffer.pop_back(); + } }; }}} //namespace diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index c7f084f62..819874483 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -335,10 +335,6 @@ libusb_zero_copy_impl::libusb_zero_copy_impl( if (send_xfer_size == 0) send_xfer_size = DEFAULT_XFER_SIZE; if (send_num_xfers == 0) send_num_xfers = DEFAULT_NUM_XFERS; - //sanity check the transfer sizes - UHD_ASSERT_THROW(recv_xfer_size % 512 == 0); - UHD_ASSERT_THROW(send_xfer_size % 512 == 0); - //store the num xfers for the num frames count _recv_xfer_size = recv_xfer_size; _recv_num_frames = recv_num_xfers; -- cgit v1.2.3 From 0cd5375b5c8a928f112a963d9c9c2556bafed108 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Tue, 5 Oct 2010 10:30:28 -0700 Subject: uhd: replaced frame params for the zero copy interfaces with a device address the device address gives a key, value pair of infinite optional capabilities added a cast option to the device address to cast string to type T added call to the zero_copy_if to get send and recv frame sizes changed the usrp2 impl to calculate recv/send spp from the data transport --- host/include/uhd/transport/udp_zero_copy.hpp | 7 ++- host/include/uhd/transport/usb_control.hpp | 2 +- host/include/uhd/transport/usb_zero_copy.hpp | 17 +++---- host/include/uhd/transport/zero_copy.hpp | 26 +++++++--- host/include/uhd/types/device_addr.hpp | 20 ++++++++ host/lib/transport/libusb1_zero_copy.cpp | 67 +++++++++++-------------- host/lib/transport/udp_zero_copy_asio.cpp | 35 +++++++------ host/lib/transport/usb_dummy_impl.cpp | 4 +- host/lib/usrp/usrp1/usrp1_impl.cpp | 48 +++++++----------- host/lib/usrp/usrp2/io_impl.cpp | 4 +- host/lib/usrp/usrp2/mboard_impl.cpp | 8 +-- host/lib/usrp/usrp2/usrp2_impl.cpp | 40 ++++++--------- host/lib/usrp/usrp2/usrp2_impl.hpp | 75 ++++++++-------------------- 13 files changed, 162 insertions(+), 191 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/include/uhd/transport/udp_zero_copy.hpp b/host/include/uhd/transport/udp_zero_copy.hpp index 818709973..bbba97b21 100644 --- a/host/include/uhd/transport/udp_zero_copy.hpp +++ b/host/include/uhd/transport/udp_zero_copy.hpp @@ -20,6 +20,7 @@ #include #include +#include #include namespace uhd{ namespace transport{ @@ -50,14 +51,12 @@ public: * * \param addr a string representing the destination address * \param port a string representing the destination port - * \param recv_buff_size size in bytes for the recv buffer, 0 for automatic - * \param send_buff_size size in bytes for the send buffer, 0 for automatic + * \param hints optional parameters to pass to the underlying transport */ static sptr make( const std::string &addr, const std::string &port, - size_t recv_buff_size = 0, - size_t send_buff_size = 0 + const device_addr_t &hints = device_addr_t() ); }; diff --git a/host/include/uhd/transport/usb_control.hpp b/host/include/uhd/transport/usb_control.hpp index f9829c3ec..e6c32f78e 100644 --- a/host/include/uhd/transport/usb_control.hpp +++ b/host/include/uhd/transport/usb_control.hpp @@ -18,7 +18,7 @@ #ifndef INCLUDED_UHD_TRANSPORT_USB_CONTROL_HPP #define INCLUDED_UHD_TRANSPORT_USB_CONTROL_HPP -#include "usb_device_handle.hpp" +#include namespace uhd { namespace transport { diff --git a/host/include/uhd/transport/usb_zero_copy.hpp b/host/include/uhd/transport/usb_zero_copy.hpp index 61bf380ba..b39171fba 100644 --- a/host/include/uhd/transport/usb_zero_copy.hpp +++ b/host/include/uhd/transport/usb_zero_copy.hpp @@ -18,8 +18,9 @@ #ifndef INCLUDED_UHD_TRANSPORT_USB_ZERO_COPY_HPP #define INCLUDED_UHD_TRANSPORT_USB_ZERO_COPY_HPP -#include "usb_device_handle.hpp" +#include #include +#include namespace uhd { namespace transport { @@ -47,19 +48,13 @@ public: * \param handle a device handle that uniquely identifying the device * \param recv_endpoint an integer specifiying an IN endpoint number * \param send_endpoint an integer specifiying an OUT endpoint number - * \param recv_xfer_size the number of bytes for each receive transfer - * \param recv_num_xfers the number of simultaneous receive transfers - * \param send_xfer_size the number of bytes for each send transfer - * \param send_num_xfers the number of simultaneous send transfers + * \param hints optional parameters to pass to the underlying transport */ static sptr make( usb_device_handle::sptr handle, - unsigned int recv_endpoint, - unsigned int send_endpoint, - size_t recv_xfer_size = 0, - size_t recv_num_xfers = 0, - size_t send_xfer_size = 0, - size_t send_num_xfers = 0 + size_t recv_endpoint, + size_t send_endpoint, + const device_addr_t &hints = device_addr_t() ); }; diff --git a/host/include/uhd/transport/zero_copy.hpp b/host/include/uhd/transport/zero_copy.hpp index 9dd16280c..7d8fb4b83 100644 --- a/host/include/uhd/transport/zero_copy.hpp +++ b/host/include/uhd/transport/zero_copy.hpp @@ -155,14 +155,19 @@ namespace uhd{ namespace transport{ virtual managed_recv_buffer::sptr get_recv_buff(double timeout = 0.1) = 0; /*! - * Get the maximum number of receive frames: - * The maximum number of valid managed recv buffers, - * or the maximum number of frames in the ring buffer, - * depending upon the underlying implementation. + * Get the number of receive frames: + * The number of simultaneous receive buffers in use. * \return number of frames */ virtual size_t get_num_recv_frames(void) const = 0; + /*! + * Get the size of a receive frame: + * The maximum capacity of a single receive buffer. + * \return frame size in bytes + */ + virtual size_t get_recv_frame_size(void) const = 0; + /*! * Get a new send buffer from this transport object. * \param timeout the timeout to get the buffer in seconds @@ -171,14 +176,19 @@ namespace uhd{ namespace transport{ virtual managed_send_buffer::sptr get_send_buff(double timeout = 0.1) = 0; /*! - * Get the maximum number of send frames: - * The maximum number of valid managed send buffers, - * or the maximum number of frames in the ring buffer, - * depending upon the underlying implementation. + * Get the number of send frames: + * The number of simultaneous send buffers in use. * \return number of frames */ virtual size_t get_num_send_frames(void) const = 0; + /*! + * Get the size of a send frame: + * The maximum capacity of a single send buffer. + * \return frame size in bytes + */ + virtual size_t get_send_frame_size(void) const = 0; + }; }} //namespace diff --git a/host/include/uhd/types/device_addr.hpp b/host/include/uhd/types/device_addr.hpp index e359d9467..eb3394230 100644 --- a/host/include/uhd/types/device_addr.hpp +++ b/host/include/uhd/types/device_addr.hpp @@ -20,6 +20,8 @@ #include #include +#include +#include #include #include @@ -62,6 +64,24 @@ namespace uhd{ * \return a string with delimiter markup */ std::string to_string(void) const; + + /*! + * Lexically cast a parameter to the specified type, + * or use the default value if the key is not found. + * \param key the key as one of the address parameters + * \param def the value to use when key is not present + * \return the casted value as type T or the default + * \throw error when the parameter cannot be casted + */ + template T cast(const std::string &key, const T &def) const{ + if (not this->has_key(key)) return def; + try{ + return boost::lexical_cast((*this)[key]); + } + catch(const boost::bad_lexical_cast &){ + throw std::runtime_error("cannot cast " + key + " = " + (*this)[key]); + } + } }; //handy typedef for a vector of device addresses diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index 819874483..df6db1eb9 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -26,6 +26,7 @@ #include #include +using namespace uhd; using namespace uhd::transport; static const double CLEANUP_TIMEOUT = 0.2; //seconds @@ -284,16 +285,19 @@ public: libusb_zero_copy_impl( libusb::device_handle::sptr handle, - unsigned int recv_endpoint, unsigned int send_endpoint, - size_t recv_xfer_size, size_t recv_num_xfers, - size_t send_xfer_size, size_t send_num_xfers + size_t recv_endpoint, + size_t send_endpoint, + const device_addr_t &hints ); managed_recv_buffer::sptr get_recv_buff(double); managed_send_buffer::sptr get_send_buff(double); - size_t get_num_recv_frames(void) const { return _recv_num_frames; } - size_t get_num_send_frames(void) const { return _send_num_frames; } + size_t get_num_recv_frames(void) const { return _num_recv_frames; } + size_t get_num_send_frames(void) const { return _num_send_frames; } + + size_t get_recv_frame_size(void) const { return _recv_frame_size; } + size_t get_send_frame_size(void) const { return _send_frame_size; } private: void release(libusb_transfer *lut){ @@ -311,8 +315,8 @@ private: } libusb::device_handle::sptr _handle; - size_t _recv_xfer_size, _send_xfer_size; - size_t _recv_num_frames, _send_num_frames; + const size_t _recv_frame_size, _num_recv_frames; + const size_t _send_frame_size, _num_send_frames; usb_endpoint::sptr _recv_ep, _send_ep; }; @@ -323,24 +327,16 @@ private: */ libusb_zero_copy_impl::libusb_zero_copy_impl( libusb::device_handle::sptr handle, - unsigned int recv_endpoint, unsigned int send_endpoint, - size_t recv_xfer_size, size_t recv_num_xfers, - size_t send_xfer_size, size_t send_num_xfers -){ - _handle = handle; - - //if the sizes are left at 0 (automatic) -> use the defaults - if (recv_xfer_size == 0) recv_xfer_size = DEFAULT_XFER_SIZE; - if (recv_num_xfers == 0) recv_num_xfers = DEFAULT_NUM_XFERS; - if (send_xfer_size == 0) send_xfer_size = DEFAULT_XFER_SIZE; - if (send_num_xfers == 0) send_num_xfers = DEFAULT_NUM_XFERS; - - //store the num xfers for the num frames count - _recv_xfer_size = recv_xfer_size; - _recv_num_frames = recv_num_xfers; - _send_xfer_size = send_xfer_size; - _send_num_frames = send_num_xfers; - + size_t recv_endpoint, + size_t send_endpoint, + const device_addr_t &hints +): + _handle(handle), + _recv_frame_size(size_t(hints.cast("recv_frame_size", DEFAULT_XFER_SIZE))), + _num_recv_frames(size_t(hints.cast("num_recv_frames", DEFAULT_NUM_XFERS))), + _send_frame_size(size_t(hints.cast("send_frame_size", DEFAULT_XFER_SIZE))), + _num_send_frames(size_t(hints.cast("num_send_frames", DEFAULT_NUM_XFERS))) +{ _handle->claim_interface(2 /*in interface*/); _handle->claim_interface(1 /*out interface*/); @@ -348,16 +344,16 @@ libusb_zero_copy_impl::libusb_zero_copy_impl( _handle, // libusb device_handle recv_endpoint, // USB endpoint number true, // IN endpoint - recv_xfer_size, // buffer size per transfer - recv_num_xfers // number of libusb transfers + this->get_recv_frame_size(), // buffer size per transfer + this->get_num_recv_frames() // number of libusb transfers )); _send_ep = usb_endpoint::sptr(new usb_endpoint( _handle, // libusb device_handle send_endpoint, // USB endpoint number false, // OUT endpoint - send_xfer_size, // buffer size per transfer - send_num_xfers // number of libusb transfers + this->get_send_frame_size(), // buffer size per transfer + this->get_num_send_frames() // number of libusb transfers )); } @@ -394,7 +390,7 @@ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){ } else { return managed_send_buffer::make_safe( - boost::asio::mutable_buffer(lut->buffer, _send_xfer_size), + boost::asio::mutable_buffer(lut->buffer, this->get_send_frame_size()), boost::bind(&libusb_zero_copy_impl::commit, shared_from_this(), lut, _1) ); } @@ -405,17 +401,14 @@ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){ **********************************************************************/ usb_zero_copy::sptr usb_zero_copy::make( usb_device_handle::sptr handle, - unsigned int recv_endpoint, unsigned int send_endpoint, - size_t recv_xfer_size, size_t recv_num_xfers, - size_t send_xfer_size, size_t send_num_xfers + size_t recv_endpoint, + size_t send_endpoint, + const device_addr_t &hints ){ libusb::device_handle::sptr dev_handle(libusb::device_handle::get_cached_handle( boost::static_pointer_cast(handle)->get_device() )); return sptr(new libusb_zero_copy_impl( - dev_handle, - recv_endpoint, send_endpoint, - recv_xfer_size, recv_num_xfers, - send_xfer_size, send_num_xfers + dev_handle, recv_endpoint, send_endpoint, hints )); } diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index e9d91fe45..ada282e07 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -44,7 +44,7 @@ static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(4 * 25e6 * 0.5); static const size_t MIN_SEND_SOCK_BUFF_SIZE = size_t(10e3); //the number of async frames to allocate for each send and recv -static const size_t DEFAULT_NUM_ASYNC_FRAMES = 32; +static const size_t DEFAULT_NUM_FRAMES = 32; //a single concurrent thread for io_service seems to be the fastest static const size_t CONCURRENCY_HINT = 1; @@ -61,14 +61,15 @@ public: typedef boost::shared_ptr sptr; udp_zero_copy_asio_impl( - const std::string &addr, const std::string &port, - size_t recv_frame_size, size_t num_recv_frames, - size_t send_frame_size, size_t num_send_frames + const std::string &addr, + const std::string &port, + const device_addr_t &hints ): - _io_service(CONCURRENCY_HINT), - _work(new asio::io_service::work(_io_service)), - _recv_frame_size(recv_frame_size), _num_recv_frames(num_recv_frames), - _send_frame_size(send_frame_size), _num_send_frames(num_send_frames) + _io_service(hints.cast("concurrency_hint", CONCURRENCY_HINT)), + _recv_frame_size(size_t(hints.cast("recv_frame_size", udp_simple::mtu))), + _num_recv_frames(size_t(hints.cast("num_recv_frames", DEFAULT_NUM_FRAMES))), + _send_frame_size(size_t(hints.cast("send_frame_size", udp_simple::mtu))), + _num_send_frames(size_t(hints.cast("num_send_frames", DEFAULT_NUM_FRAMES))) { //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl; @@ -99,6 +100,7 @@ public: } //spawn the service threads that will run the io service + _work = new asio::io_service::work(_io_service); //new work to delete later for (size_t i = 0; i < CONCURRENCY_HINT; i++) _thread_group.create_thread( boost::bind(&udp_zero_copy_asio_impl::service, this) ); @@ -141,6 +143,7 @@ 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;} //! pop an empty send buffer off of the fifo and bind with the commit callback managed_send_buffer::sptr get_send_buff(double timeout){ @@ -159,6 +162,7 @@ public: } size_t get_num_send_frames(void) const {return _num_send_frames;} + size_t get_send_frame_size(void) const {return _send_frame_size;} private: void service(void){ @@ -260,14 +264,15 @@ template static void resize_buff_helper( udp_zero_copy::sptr udp_zero_copy::make( const std::string &addr, const std::string &port, - size_t recv_buff_size, - size_t send_buff_size + const device_addr_t &hints ){ - udp_zero_copy_asio_impl::sptr udp_trans(new udp_zero_copy_asio_impl( - addr, port, - udp_simple::mtu, DEFAULT_NUM_ASYNC_FRAMES, //recv - udp_simple::mtu, DEFAULT_NUM_ASYNC_FRAMES //send - )); + udp_zero_copy_asio_impl::sptr udp_trans( + new udp_zero_copy_asio_impl(addr, port, hints) + ); + + //extract buffer size hints from the device addr + size_t recv_buff_size = size_t(hints.cast("recv_buff_size", 0.0)); + size_t send_buff_size = size_t(hints.cast("send_buff_size", 0.0)); //call the helper to resize send and recv buffers resize_buff_helper(udp_trans, recv_buff_size, "recv"); diff --git a/host/lib/transport/usb_dummy_impl.cpp b/host/lib/transport/usb_dummy_impl.cpp index 518342aba..8a9772e7f 100644 --- a/host/lib/transport/usb_dummy_impl.cpp +++ b/host/lib/transport/usb_dummy_impl.cpp @@ -20,6 +20,7 @@ #include #include +using namespace uhd; using namespace uhd::transport; std::vector usb_device_handle::get_device_list(boost::uint16_t, boost::uint16_t){ @@ -32,8 +33,7 @@ usb_control::sptr usb_control::make(usb_device_handle::sptr){ usb_zero_copy::sptr usb_zero_copy::make( usb_device_handle::sptr, - unsigned int, unsigned int, - size_t, size_t, size_t, size_t + size_t, size_t, const device_addr_t & ){ throw std::runtime_error("no usb support -> usb_zero_copy::make not implemented"); } diff --git a/host/lib/usrp/usrp1/usrp1_impl.cpp b/host/lib/usrp/usrp1/usrp1_impl.cpp index 6ebd6bb09..276ca86f6 100644 --- a/host/lib/usrp/usrp1/usrp1_impl.cpp +++ b/host/lib/usrp/usrp1/usrp1_impl.cpp @@ -106,17 +106,8 @@ static device_addrs_t usrp1_find(const device_addr_t &hint) /*********************************************************************** * Make **********************************************************************/ -template static output_type cast_from_dev_addr( - const device_addr_t &device_addr, - const std::string &key, - output_type def_val -){ - return (device_addr.has_key(key))? - boost::lexical_cast(device_addr[key]) : def_val; -} +static device::sptr usrp1_make(const device_addr_t &device_addr){ -static device::sptr usrp1_make(const device_addr_t &device_addr) -{ //extract the FPGA path for the USRP1 std::string usrp1_fpga_image = find_image_path( device_addr.has_key("fpga")? device_addr["fpga"] : "usrp1_fpga.rbf" @@ -127,29 +118,26 @@ static device::sptr usrp1_make(const device_addr_t &device_addr) std::vector device_list = usb_device_handle::get_device_list(USRP1_VENDOR_ID, USRP1_PRODUCT_ID); - //create data and control transports - usb_zero_copy::sptr data_transport; - usrp_ctrl::sptr usrp_ctrl; - - - BOOST_FOREACH(usb_device_handle::sptr handle, device_list) { - if (handle->get_serial() == device_addr["serial"]) { - usb_control::sptr ctrl_transport = usb_control::make(handle); - usrp_ctrl = usrp_ctrl::make(ctrl_transport); - usrp_ctrl->usrp_load_fpga(usrp1_fpga_image); - - data_transport = usb_zero_copy::make( - handle, // identifier - 6, // IN endpoint - 2, // OUT endpoint - size_t(cast_from_dev_addr(device_addr, "recv_xfer_size", 0)), - size_t(cast_from_dev_addr(device_addr, "recv_num_xfers", 0)), - size_t(cast_from_dev_addr(device_addr, "send_xfer_size", 0)), - size_t(cast_from_dev_addr(device_addr, "send_num_xfers", 0)) - ); + //locate the matching handle in the device list + usb_device_handle::sptr handle; + BOOST_FOREACH(usb_device_handle::sptr dev_handle, device_list) { + if (dev_handle->get_serial() == device_addr["serial"]){ + handle = dev_handle; break; } } + UHD_ASSERT_THROW(handle.get() != NULL); //better be found + + //create control objects and a data transport + usb_control::sptr ctrl_transport = usb_control::make(handle); + usrp_ctrl::sptr usrp_ctrl = usrp_ctrl::make(ctrl_transport); + usrp_ctrl->usrp_load_fpga(usrp1_fpga_image); + usb_zero_copy::sptr data_transport = usb_zero_copy::make( + handle, // identifier + 6, // IN endpoint + 2, // OUT endpoint + device_addr // param hints + ); //create the usrp1 implementation guts return device::sptr(new usrp1_impl(data_transport, usrp_ctrl)); diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp index 33cec3cbc..07eda08c3 100644 --- a/host/lib/usrp/usrp2/io_impl.cpp +++ b/host/lib/usrp/usrp2/io_impl.cpp @@ -206,7 +206,7 @@ size_t usrp2_impl::send( _io_impl->packet_handler_send_state, //last state of the send handler buffs, num_samps, //buffer to fill metadata, send_mode, //samples metadata - io_type, _io_helper.get_tx_otw_type(), //input and output types to convert + 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(&get_send_buffs, _data_transports, _1, timeout), @@ -226,7 +226,7 @@ size_t usrp2_impl::recv( _io_impl->packet_handler_recv_state, //last state of the recv handler buffs, num_samps, //buffer to fill metadata, recv_mode, //samples metadata - io_type, _io_helper.get_rx_otw_type(), //input and output types to convert + 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(), _1, timeout) diff --git a/host/lib/usrp/usrp2/mboard_impl.cpp b/host/lib/usrp/usrp2/mboard_impl.cpp index 0b9f8ee83..a0e6adfad 100644 --- a/host/lib/usrp/usrp2/mboard_impl.cpp +++ b/host/lib/usrp/usrp2/mboard_impl.cpp @@ -38,10 +38,10 @@ using namespace uhd::usrp; usrp2_mboard_impl::usrp2_mboard_impl( size_t index, transport::udp_simple::sptr ctrl_transport, - const usrp2_io_helper &io_helper + size_t recv_frame_size ): _index(index), - _io_helper(io_helper) + _recv_frame_size(recv_frame_size) { //make a new interface for usrp2 stuff _iface = usrp2_iface::make(ctrl_transport); @@ -75,7 +75,7 @@ usrp2_mboard_impl::usrp2_mboard_impl( this->issue_ddc_stream_cmd(stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS); //init the rx control registers - _iface->poke32(U2_REG_RX_CTRL_NSAMPS_PER_PKT, _io_helper.get_max_recv_samps_per_packet()); + _iface->poke32(U2_REG_RX_CTRL_NSAMPS_PER_PKT, _recv_frame_size); _iface->poke32(U2_REG_RX_CTRL_NCHANNELS, 1); _iface->poke32(U2_REG_RX_CTRL_CLEAR_OVERRUN, 1); //reset _iface->poke32(U2_REG_RX_CTRL_VRT_HEADER, 0 @@ -178,7 +178,7 @@ void usrp2_mboard_impl::set_time_spec(const time_spec_t &time_spec, bool now){ void usrp2_mboard_impl::issue_ddc_stream_cmd(const stream_cmd_t &stream_cmd){ _iface->poke32(U2_REG_RX_CTRL_STREAM_CMD, dsp_type1::calc_stream_cmd_word( - stream_cmd, _io_helper.get_max_recv_samps_per_packet() + stream_cmd, _recv_frame_size )); _iface->poke32(U2_REG_RX_CTRL_TIME_SECS, boost::uint32_t(stream_cmd.time_spec.get_full_secs())); _iface->poke32(U2_REG_RX_CTRL_TIME_TICKS, stream_cmd.time_spec.get_tick_count(get_master_clock_freq())); diff --git a/host/lib/usrp/usrp2/usrp2_impl.cpp b/host/lib/usrp/usrp2/usrp2_impl.cpp index 568c87a22..a680708ad 100644 --- a/host/lib/usrp/usrp2/usrp2_impl.cpp +++ b/host/lib/usrp/usrp2/usrp2_impl.cpp @@ -124,26 +124,7 @@ static uhd::device_addrs_t usrp2_find(const device_addr_t &hint){ /*********************************************************************** * Make **********************************************************************/ -template -out_type lexical_cast(const in_type &in){ - try{ - return boost::lexical_cast(in); - }catch(...){ - throw std::runtime_error(str(boost::format( - "failed to cast \"%s\" to type \"%s\"" - ) % boost::lexical_cast(in) % typeid(out_type).name())); - } -} - static device::sptr usrp2_make(const device_addr_t &device_addr){ - //extract the receive and send buffer sizes - size_t recv_buff_size = 0, send_buff_size= 0 ; - if (device_addr.has_key("recv_buff_size")){ - recv_buff_size = size_t(lexical_cast(device_addr["recv_buff_size"])); - } - if (device_addr.has_key("send_buff_size")){ - send_buff_size = size_t(lexical_cast(device_addr["send_buff_size"])); - } //create a ctrl and data transport for each address std::vector ctrl_transports; @@ -154,8 +135,7 @@ static device::sptr usrp2_make(const device_addr_t &device_addr){ addr, num2str(USRP2_UDP_CTRL_PORT) )); data_transports.push_back(udp_zero_copy::make( - addr, num2str(USRP2_UDP_DATA_PORT), - recv_buff_size, send_buff_size + addr, num2str(USRP2_UDP_DATA_PORT), device_addr )); } @@ -178,11 +158,23 @@ usrp2_impl::usrp2_impl( ): _data_transports(data_transports) { + //setup rx otw type + _rx_otw_type.width = 16; + _rx_otw_type.shift = 0; + _rx_otw_type.byteorder = uhd::otw_type_t::BO_BIG_ENDIAN; + + //setup tx otw type + _tx_otw_type.width = 16; + _tx_otw_type.shift = 0; + _tx_otw_type.byteorder = uhd::otw_type_t::BO_BIG_ENDIAN; + + //!!!!! set the otw type here before continuing, its used below + //create a new mboard handler for each control transport for(size_t i = 0; i < ctrl_transports.size(); i++){ - _mboards.push_back(usrp2_mboard_impl::sptr( - new usrp2_mboard_impl(i, ctrl_transports[i], _io_helper) - )); + _mboards.push_back(usrp2_mboard_impl::sptr(new usrp2_mboard_impl( + i, ctrl_transports[i], this->get_max_recv_samps_per_packet() + ))); //use an empty name when there is only one mboard std::string name = (ctrl_transports.size() > 1)? boost::lexical_cast(i) : ""; _mboard_dict[name] = _mboards.back(); diff --git a/host/lib/usrp/usrp2/usrp2_impl.hpp b/host/lib/usrp/usrp2/usrp2_impl.hpp index e8763b284..e12c4d6d4 100644 --- a/host/lib/usrp/usrp2/usrp2_impl.hpp +++ b/host/lib/usrp/usrp2/usrp2_impl.hpp @@ -71,54 +71,6 @@ private: void set(const wax::obj &key, const wax::obj &val){return _set(key, val);} }; -/*! - * The io helper class encapculates the max packet sizes and otw types. - * The otw types are read-only for now, this will be reimplemented - * when it becomes possible to change the otw type in the usrp2. - */ -class usrp2_io_helper{ -public: - usrp2_io_helper(void){ - //setup rx otw type - _rx_otw_type.width = 16; - _rx_otw_type.shift = 0; - _rx_otw_type.byteorder = uhd::otw_type_t::BO_BIG_ENDIAN; - - //setup tx otw type - _tx_otw_type.width = 16; - _tx_otw_type.shift = 0; - _tx_otw_type.byteorder = uhd::otw_type_t::BO_BIG_ENDIAN; - } - - inline size_t get_max_send_samps_per_packet(void) const{ - return _max_tx_bytes_per_packet/_tx_otw_type.get_sample_size(); - } - - inline size_t get_max_recv_samps_per_packet(void) const{ - return _max_rx_bytes_per_packet/_rx_otw_type.get_sample_size(); - } - - inline const uhd::otw_type_t &get_rx_otw_type(void) const{ - return _rx_otw_type; - } - - inline const uhd::otw_type_t &get_tx_otw_type(void) const{ - return _tx_otw_type; - } - -private: - uhd::otw_type_t _rx_otw_type, _tx_otw_type; - static const size_t _max_rx_bytes_per_packet = uhd::transport::udp_simple::mtu - - uhd::transport::vrt::max_if_hdr_words32*sizeof(boost::uint32_t) - - sizeof(uhd::transport::vrt::if_packet_info_t().tlr) //forced to have trailer - + sizeof(uhd::transport::vrt::if_packet_info_t().cid) //no class id ever used - ; - static const size_t _max_tx_bytes_per_packet = uhd::transport::udp_simple::mtu - - uhd::transport::vrt::max_if_hdr_words32*sizeof(boost::uint32_t) - + sizeof(uhd::transport::vrt::if_packet_info_t().cid) //no class id ever used - ; -}; - /*! * USRP2 mboard implementation guts: * The implementation details are encapsulated here. @@ -129,7 +81,11 @@ public: typedef boost::shared_ptr sptr; //structors - usrp2_mboard_impl(size_t index, uhd::transport::udp_simple::sptr, const usrp2_io_helper &); + usrp2_mboard_impl( + size_t index, + uhd::transport::udp_simple::sptr, + size_t recv_frame_size + ); ~usrp2_mboard_impl(void); inline double get_master_clock_freq(void){ @@ -139,7 +95,7 @@ public: private: size_t _index; int _rev_hi, _rev_lo; - const usrp2_io_helper &_io_helper; + const size_t _recv_frame_size; //properties for this mboard void get(const wax::obj &, wax::obj &); @@ -229,7 +185,8 @@ public: //the io interface size_t get_max_send_samps_per_packet(void) const{ - return _io_helper.get_max_send_samps_per_packet(); + const size_t bytes_per_packet = _data_transports.front()->get_send_frame_size() - _max_tx_header_bytes; + return bytes_per_packet/_tx_otw_type.get_sample_size(); } size_t send( const std::vector &, size_t, @@ -237,7 +194,8 @@ public: uhd::device::send_mode_t, double ); size_t get_max_recv_samps_per_packet(void) const{ - return _io_helper.get_max_recv_samps_per_packet(); + const size_t bytes_per_packet = _data_transports.front()->get_recv_frame_size() - _max_rx_header_bytes; + return bytes_per_packet/_rx_otw_type.get_sample_size(); } size_t recv( const std::vector &, size_t, @@ -257,9 +215,20 @@ private: //io impl methods and members std::vector _data_transports; - const usrp2_io_helper _io_helper; UHD_PIMPL_DECL(io_impl) _io_impl; void io_init(void); + + //over-the-wire structs and constants + uhd::otw_type_t _rx_otw_type, _tx_otw_type; + static const size_t _max_rx_header_bytes = 0 + + uhd::transport::vrt::max_if_hdr_words32*sizeof(boost::uint32_t) + + sizeof(uhd::transport::vrt::if_packet_info_t().tlr) //forced to have trailer + - sizeof(uhd::transport::vrt::if_packet_info_t().cid) //no class id ever used + ; + static const size_t _max_tx_header_bytes = 0 + + uhd::transport::vrt::max_if_hdr_words32*sizeof(boost::uint32_t) + - sizeof(uhd::transport::vrt::if_packet_info_t().cid) //no class id ever used + ; }; #endif /* INCLUDED_USRP2_IMPL_HPP */ -- cgit v1.2.3 From 0fc641042ef81731fa84f796fd983a5a602e260c Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Tue, 5 Oct 2010 15:47:58 -0700 Subject: usb: moved event handler thread into the zero copy interface --- host/docs/transport.rst | 1 + host/lib/transport/libusb1_base.cpp | 18 ------------------ host/lib/transport/libusb1_zero_copy.cpp | 28 ++++++++++++++++++++++++++++ 3 files changed, 29 insertions(+), 18 deletions(-) (limited to 'host/lib/transport/libusb1_zero_copy.cpp') diff --git a/host/docs/transport.rst b/host/docs/transport.rst index d6a146c67..30fc1d78f 100644 --- a/host/docs/transport.rst +++ b/host/docs/transport.rst @@ -79,3 +79,4 @@ The following parameters can be used to alter the transport's default behavior: * **num_recv_frames:** The number of simultaneous receive transfers * **send_frame_size:** The size of a single send transfers in bytes * **num_send_frames:** The number of simultaneous send transfers +* **concurrency_hint:** The number of threads to run the event handler diff --git a/host/lib/transport/libusb1_base.cpp b/host/lib/transport/libusb1_base.cpp index 910b04fc8..cfa77d9ca 100644 --- a/host/lib/transport/libusb1_base.cpp +++ b/host/lib/transport/libusb1_base.cpp @@ -16,12 +16,10 @@ // #include "libusb1_base.hpp" -#include #include #include #include #include -#include #include using namespace uhd; @@ -35,12 +33,9 @@ public: libusb_session_impl(void){ UHD_ASSERT_THROW(libusb_init(&_context) == 0); libusb_set_debug(_context, debug_level); - _thread_group.create_thread(boost::bind(&libusb_session_impl::run_event_loop, this)); } ~libusb_session_impl(void){ - _running = false; - _thread_group.join_all(); libusb_exit(_context); } @@ -50,19 +45,6 @@ public: private: libusb_context *_context; - boost::thread_group _thread_group; - bool _running; - - void run_event_loop(void){ - set_thread_priority_safe(); - _running = true; - timeval tv; - while(_running){ - tv.tv_sec = 0; - tv.tv_usec = 100000; //100ms - libusb_handle_events_timeout(this->get_context(), &tv); - } - } }; libusb::session::sptr libusb::session::get_global_session(void){ diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp index df6db1eb9..f589d7c77 100644 --- a/host/lib/transport/libusb1_zero_copy.cpp +++ b/host/lib/transport/libusb1_zero_copy.cpp @@ -18,6 +18,7 @@ #include "libusb1_base.hpp" #include #include +#include #include #include #include @@ -290,6 +291,11 @@ public: const device_addr_t &hints ); + ~libusb_zero_copy_impl(void){ + _threads_running = false; + _thread_group.join_all(); + } + managed_recv_buffer::sptr get_recv_buff(double); managed_send_buffer::sptr get_send_buff(double); @@ -318,6 +324,22 @@ private: const size_t _recv_frame_size, _num_recv_frames; const size_t _send_frame_size, _num_send_frames; usb_endpoint::sptr _recv_ep, _send_ep; + + //event handler threads + boost::thread_group _thread_group; + bool _threads_running; + + void run_event_loop(void){ + set_thread_priority_safe(); + libusb::session::sptr session = libusb::session::get_global_session(); + _threads_running = true; + while(_threads_running){ + timeval tv; + tv.tv_sec = 0; + tv.tv_usec = 100000; //100ms + libusb_handle_events_timeout(session->get_context(), &tv); + } + } }; /* @@ -355,6 +377,12 @@ libusb_zero_copy_impl::libusb_zero_copy_impl( this->get_send_frame_size(), // buffer size per transfer this->get_num_send_frames() // number of libusb transfers )); + + //spawn the event handler threads + size_t concurrency = hints.cast("concurrency_hint", 1); + for (size_t i = 0; i < concurrency; i++) _thread_group.create_thread( + boost::bind(&libusb_zero_copy_impl::run_event_loop, this) + ); } /* -- cgit v1.2.3