From f2f4840ff9e15ed21344a5a19d676d548b542e91 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Fri, 24 Sep 2010 11:07:14 -0700 Subject: usb: moved msvc stdint file and changed cmake conditional for msvc --- host/lib/transport/CMakeLists.txt | 6 +++--- host/lib/transport/include/stdint.h | 35 ----------------------------------- host/lib/transport/msvc/stdint.h | 35 +++++++++++++++++++++++++++++++++++ 3 files changed, 38 insertions(+), 38 deletions(-) delete mode 100644 host/lib/transport/include/stdint.h create mode 100644 host/lib/transport/msvc/stdint.h (limited to 'host/lib/transport') diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt index 92c9f3181..48b6e9677 100644 --- a/host/lib/transport/CMakeLists.txt +++ b/host/lib/transport/CMakeLists.txt @@ -33,9 +33,9 @@ IF(LIBUSB_FOUND) ${CMAKE_SOURCE_DIR}/lib/transport/libusb1_base.hpp ${CMAKE_SOURCE_DIR}/lib/transport/libusb1_device_handle.cpp ) - IF(WIN32) #include our custom stdint for libusb - INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/lib/transport/include) - ENDIF(WIN32) + IF(MSVC) #include our custom stdint for libusb + INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/lib/transport/msvc) + ENDIF(MSVC) SET(HAVE_USB_SUPPORT TRUE) ENDIF(LIBUSB_FOUND) diff --git a/host/lib/transport/include/stdint.h b/host/lib/transport/include/stdint.h deleted file mode 100644 index b3eb61aae..000000000 --- a/host/lib/transport/include/stdint.h +++ /dev/null @@ -1,35 +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 . -// - -#ifndef INCLUDED_LIBUHD_TRANSPORT_STDINT_H -#define INCLUDED_LIBUHD_TRANSPORT_STDINT_H - -#include - -//provide a stdint implementation for libusb - -typedef boost::uint64_t uint64_t; -typedef boost::uint32_t uint32_t; -typedef boost::uint16_t uint16_t; -typedef boost::uint8_t uint8_t; - -typedef boost::int64_t int64_t; -typedef boost::int32_t int32_t; -typedef boost::int16_t int16_t; -typedef boost::int8_t int8_t; - -#endif /* INCLUDED_LIBUHD_TRANSPORT_STDINT_H */ diff --git a/host/lib/transport/msvc/stdint.h b/host/lib/transport/msvc/stdint.h new file mode 100644 index 000000000..b3eb61aae --- /dev/null +++ b/host/lib/transport/msvc/stdint.h @@ -0,0 +1,35 @@ +// +// 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 . +// + +#ifndef INCLUDED_LIBUHD_TRANSPORT_STDINT_H +#define INCLUDED_LIBUHD_TRANSPORT_STDINT_H + +#include + +//provide a stdint implementation for libusb + +typedef boost::uint64_t uint64_t; +typedef boost::uint32_t uint32_t; +typedef boost::uint16_t uint16_t; +typedef boost::uint8_t uint8_t; + +typedef boost::int64_t int64_t; +typedef boost::int32_t int32_t; +typedef boost::int16_t int16_t; +typedef boost::int8_t int8_t; + +#endif /* INCLUDED_LIBUHD_TRANSPORT_STDINT_H */ -- cgit v1.2.3 From 1dffd5bb5fe16cd35ba54e44e927888e0192e905 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Fri, 24 Sep 2010 18:14:32 -0700 Subject: usb: tweaks to usb code to cleanup properly and/or in error conditions --- host/lib/transport/libusb1_base.cpp | 17 +++++++---------- host/lib/transport/libusb1_control.cpp | 1 + host/lib/transport/libusb1_device_handle.cpp | 9 +++++---- 3 files changed, 13 insertions(+), 14 deletions(-) (limited to 'host/lib/transport') diff --git a/host/lib/transport/libusb1_base.cpp b/host/lib/transport/libusb1_base.cpp index 1f816c6e2..cd3e4adcf 100644 --- a/host/lib/transport/libusb1_base.cpp +++ b/host/lib/transport/libusb1_base.cpp @@ -30,9 +30,7 @@ using namespace uhd::transport; **********************************************************/ void libusb::init(libusb_context **ctx, int debug_level) { - if (libusb_init(ctx) < 0) - std::cerr << "error: libusb_init" << std::endl; - + UHD_ASSERT_THROW(libusb_init(ctx) == 0); libusb_set_debug(*ctx, debug_level); } @@ -48,14 +46,13 @@ libusb_device_handle *libusb::open_device(libusb_context *ctx, libusb_device *dev = libusb_dev_list[i]; if (compare_device(dev, handle)) { - libusb_open(dev, &dev_handle); - libusb_unref_device(dev); - break; + if (libusb_open(dev, &dev_handle) == 0) break; } - - libusb_unref_device(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; } @@ -70,12 +67,12 @@ bool libusb::compare_device(libusb_device *dev, boost::uint16_t device_addr = handle->get_device_addr(); libusb_device_descriptor libusb_desc; - if (libusb_get_device_descriptor(dev, &libusb_desc) < 0) + 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; + return false; if (serial != get_serial(dev)) return false; if (device_addr != libusb_get_device_address(dev)) diff --git a/host/lib/transport/libusb1_control.cpp b/host/lib/transport/libusb1_control.cpp index 3531128b2..be9b51111 100644 --- a/host/lib/transport/libusb1_control.cpp +++ b/host/lib/transport/libusb1_control.cpp @@ -46,6 +46,7 @@ private: 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 diff --git a/host/lib/transport/libusb1_device_handle.cpp b/host/lib/transport/libusb1_device_handle.cpp index 7efddd410..6bef37ed2 100644 --- a/host/lib/transport/libusb1_device_handle.cpp +++ b/host/lib/transport/libusb1_device_handle.cpp @@ -77,7 +77,7 @@ usb_device_handle::sptr make_usb_device_handle(libusb_device *dev) libusb_device_descriptor desc; if (libusb_get_device_descriptor(dev, &desc) < 0) { - UHD_ASSERT_THROW("USB: failed to get device descriptor"); + throw std::runtime_error("USB: failed to get device descriptor"); } std::string serial = libusb::get_serial(dev); @@ -104,14 +104,15 @@ std::vector usb_device_handle::get_device_list(boost::u 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) { - UHD_ASSERT_THROW("USB: failed to get device descriptor"); + 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)); + 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; } -- 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') 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') 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') 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 1180615fe2ee1e976aeb9139eb1378fb93befb0f Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sun, 26 Sep 2010 22:45:56 -0700 Subject: usb: added some documentation to wrapper base --- host/lib/transport/libusb1_base.hpp | 7 +++++++ 1 file changed, 7 insertions(+) (limited to 'host/lib/transport') diff --git a/host/lib/transport/libusb1_base.hpp b/host/lib/transport/libusb1_base.hpp index 47d078cde..64a1442d1 100644 --- a/host/lib/transport/libusb1_base.hpp +++ b/host/lib/transport/libusb1_base.hpp @@ -24,6 +24,13 @@ #include #include +/*********************************************************************** + * Libusb object oriented smart pointer wrappers: + * The following wrappers provide allocation and automatic deallocation + * for various libusb data types and handles. The construction routines + * also store tables of already allocated structures to avoid multiple + * occurrences of opened handles (for example). + **********************************************************************/ namespace uhd { namespace transport { namespace libusb { -- cgit v1.2.3 From 694404694515e293da6b8949e74af5f469a2e1e5 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Mon, 27 Sep 2010 00:32:46 -0700 Subject: usb: use the proper libusb include (in path set by pkgconfig) --- host/lib/transport/libusb1_base.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'host/lib/transport') diff --git a/host/lib/transport/libusb1_base.hpp b/host/lib/transport/libusb1_base.hpp index 64a1442d1..04c1d6574 100644 --- a/host/lib/transport/libusb1_base.hpp +++ b/host/lib/transport/libusb1_base.hpp @@ -22,7 +22,7 @@ #include #include #include -#include +#include /*********************************************************************** * Libusb object oriented smart pointer wrappers: -- 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') 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 e4c6f42ff30aeaedf2d1eb426a221b85236fb74e Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Tue, 28 Sep 2010 16:53:52 -0700 Subject: usb: submit should return ssize_t, usrp1: set hash before reset after fw load --- host/include/uhd/transport/usb_control.hpp | 4 ++-- host/lib/transport/libusb1_control.cpp | 2 +- host/lib/usrp/usrp1/usrp1_ctrl.cpp | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) (limited to 'host/lib/transport') diff --git a/host/include/uhd/transport/usb_control.hpp b/host/include/uhd/transport/usb_control.hpp index 6137ecf84..f9829c3ec 100644 --- a/host/include/uhd/transport/usb_control.hpp +++ b/host/include/uhd/transport/usb_control.hpp @@ -50,9 +50,9 @@ public: * \param index 2-byte (wIndex) * \param buff buffer to hold send or receive data * \param length 2-byte (wLength) - * \return number of bytes submitted + * \return number of bytes submitted or error code */ - virtual size_t submit(boost::uint8_t request_type, + virtual ssize_t submit(boost::uint8_t request_type, boost::uint8_t request, boost::uint16_t value, boost::uint16_t index, diff --git a/host/lib/transport/libusb1_control.cpp b/host/lib/transport/libusb1_control.cpp index c989a788c..f903907d0 100644 --- a/host/lib/transport/libusb1_control.cpp +++ b/host/lib/transport/libusb1_control.cpp @@ -33,7 +33,7 @@ public: _handle->claim_interface(0 /* control interface */); } - size_t submit(boost::uint8_t request_type, + ssize_t submit(boost::uint8_t request_type, boost::uint8_t request, boost::uint16_t value, boost::uint16_t index, diff --git a/host/lib/usrp/usrp1/usrp1_ctrl.cpp b/host/lib/usrp/usrp1/usrp1_ctrl.cpp index 2e1bd85cd..5d73e6dd9 100644 --- a/host/lib/usrp/usrp1/usrp1_ctrl.cpp +++ b/host/lib/usrp/usrp1/usrp1_ctrl.cpp @@ -209,9 +209,9 @@ public: } //type 0x01 is end else if (type == 0x01) { + usrp_set_firmware_hash(hash); //set hash before reset usrp_control_write(FX2_FIRMWARE_LOAD, 0xe600, 0, &reset_n, 1); - usrp_set_firmware_hash(hash); file.close(); //wait for things to settle -- 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') 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 e4fffed05dda57bb37d693a3a26ea6a903c925f7 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Thu, 30 Sep 2010 11:32:21 -0700 Subject: usb: set rt thread priority for the libusb event loop --- host/lib/transport/libusb1_base.cpp | 3 +++ 1 file changed, 3 insertions(+) (limited to 'host/lib/transport') diff --git a/host/lib/transport/libusb1_base.cpp b/host/lib/transport/libusb1_base.cpp index 26e864459..49f524a32 100644 --- a/host/lib/transport/libusb1_base.cpp +++ b/host/lib/transport/libusb1_base.cpp @@ -16,6 +16,7 @@ // #include "libusb1_base.hpp" +#include #include #include #include @@ -23,6 +24,7 @@ #include #include +using namespace uhd; using namespace uhd::transport; /*********************************************************************** @@ -52,6 +54,7 @@ private: bool _running; void run_event_loop(void){ + set_thread_priority_safe(); _running = true; timeval tv; while(_running){ -- 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') 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 348b5b12ad9ec916c92d642bfb28e47264473535 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Fri, 1 Oct 2010 11:54:22 -0700 Subject: usb: catch open errors and print message, device: catch exceptions at discovery time --- host/lib/device.cpp | 18 ++++++++++++------ host/lib/transport/libusb1_base.cpp | 16 +++++++++++----- host/lib/usrp/usrp1/usrp1_impl.cpp | 28 +++++++++++++--------------- 3 files changed, 36 insertions(+), 26 deletions(-) (limited to 'host/lib/transport') diff --git a/host/lib/device.cpp b/host/lib/device.cpp index d575ebaab..386588a08 100644 --- a/host/lib/device.cpp +++ b/host/lib/device.cpp @@ -26,6 +26,7 @@ #include #include #include +#include using namespace uhd; @@ -73,12 +74,17 @@ device_addrs_t device::find(const device_addr_t &hint){ device_addrs_t device_addrs; BOOST_FOREACH(const dev_fcn_reg_t &fcn, get_dev_fcn_regs()){ - device_addrs_t discovered_addrs = fcn.get<0>()(hint); - device_addrs.insert( - device_addrs.begin(), - discovered_addrs.begin(), - discovered_addrs.end() - ); + try{ + device_addrs_t discovered_addrs = fcn.get<0>()(hint); + device_addrs.insert( + device_addrs.begin(), + discovered_addrs.begin(), + discovered_addrs.end() + ); + } + catch(const std::exception &e){ + std::cerr << "Device discovery error: " << e.what() << std::endl; + } } return device_addrs; diff --git a/host/lib/transport/libusb1_base.cpp b/host/lib/transport/libusb1_base.cpp index 49f524a32..910b04fc8 100644 --- a/host/lib/transport/libusb1_base.cpp +++ b/host/lib/transport/libusb1_base.cpp @@ -213,15 +213,21 @@ private: libusb::device_handle::sptr libusb::device_handle::get_cached_handle(device::sptr dev){ static uhd::dict > handles; - //not expired -> get existing session + //not expired -> get existing handle if (handles.has_key(dev->get()) and not handles[dev->get()].expired()){ return handles[dev->get()].lock(); } - //create a new global session - sptr new_handle(new libusb_device_handle_impl(dev)); - handles[dev->get()] = new_handle; - return new_handle; + //create a new cached handle + try{ + sptr new_handle(new libusb_device_handle_impl(dev)); + handles[dev->get()] = new_handle; + return new_handle; + } + catch(const std::exception &e){ + std::cerr << "USB open failed: see the application notes for your device." << std::endl; + throw std::runtime_error(e.what()); + } } /*********************************************************************** diff --git a/host/lib/usrp/usrp1/usrp1_impl.cpp b/host/lib/usrp/usrp1/usrp1_impl.cpp index 793e3027d..156fc119f 100644 --- a/host/lib/usrp/usrp1/usrp1_impl.cpp +++ b/host/lib/usrp/usrp1/usrp1_impl.cpp @@ -83,9 +83,7 @@ static device_addrs_t usrp1_find(const device_addr_t &hint) //find the usrps and load firmware BOOST_FOREACH(usb_device_handle::sptr handle, usb_device_handle::get_device_list(vid, pid)) { - usb_control::sptr ctrl_transport = usb_control::make(handle); - usrp_ctrl::sptr usrp_ctrl = usrp_ctrl::make(ctrl_transport); - usrp_ctrl->usrp_load_firmware(usrp1_fw_image); + usrp_ctrl::make(usb_control::make(handle))->usrp_load_firmware(usrp1_fw_image); } //get descriptors again with serial number, but using the initialized VID/PID now since we have firmware @@ -93,13 +91,13 @@ static device_addrs_t usrp1_find(const device_addr_t &hint) pid = USRP1_PRODUCT_ID; BOOST_FOREACH(usb_device_handle::sptr handle, usb_device_handle::get_device_list(vid, pid)) { - device_addr_t new_addr; - new_addr["type"] = "usrp1"; - new_addr["serial"] = handle->get_serial(); - //this is a found usrp1 when a hint serial is not specified or it matches - if (not hint.has_key("serial") or hint["serial"] == new_addr["serial"]){ - usrp1_addrs.push_back(new_addr); - } + device_addr_t new_addr; + new_addr["type"] = "usrp1"; + new_addr["serial"] = handle->get_serial(); + //this is a found usrp1 when a hint serial is not specified or it matches + if (not hint.has_key("serial") or hint["serial"] == new_addr["serial"]){ + usrp1_addrs.push_back(new_addr); + } } return usrp1_addrs; @@ -109,12 +107,12 @@ 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 + 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; + 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) -- 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') 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') 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') 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') 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 57ad303e850f143573d8c6edb1a88542b7b43350 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Mon, 4 Oct 2010 09:28:26 -0700 Subject: udp: added io service work to keep service running --- host/lib/transport/udp_zero_copy_asio.cpp | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) (limited to 'host/lib/transport') diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index a1eb516fc..2cf7bde18 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -45,6 +45,9 @@ 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; +//a single concurrent thread for io_service seems to be the fastest +static const size_t CONCURRENCY_HINT = 1; + /*********************************************************************** * Zero Copy UDP implementation with ASIO: * This is the portable zero copy implementation for systems @@ -61,6 +64,8 @@ public: size_t recv_frame_size, size_t num_recv_frames, size_t send_frame_size, size_t num_send_frames ): + _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) { @@ -92,13 +97,15 @@ public: 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)); + //spawn the service threads that will run the io service + for (size_t i = 0; i < CONCURRENCY_HINT; i++) _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 _work; //allow io_service run to complete + _thread_group.join_all(); //wait for service threads to exit delete _socket; } @@ -200,6 +207,7 @@ private: //asio guts -> socket and service asio::ip::udp::socket *_socket; asio::io_service _io_service; + asio::io_service::work *_work; //memory management -> buffers and fifos boost::thread_group _thread_group; -- 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') 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') 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 50ff7e4d3bf4e196b4c054821f67f8d1bff5203b Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Mon, 4 Oct 2010 16:34:24 -0700 Subject: usb: added dummy usb implementation for building without usb (throw, not segfault) --- host/lib/transport/CMakeLists.txt | 4 +++- host/lib/transport/usb_dummy_impl.cpp | 39 +++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) create mode 100644 host/lib/transport/usb_dummy_impl.cpp (limited to 'host/lib/transport') diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt index bf92b0822..b9ec7a6ad 100644 --- a/host/lib/transport/CMakeLists.txt +++ b/host/lib/transport/CMakeLists.txt @@ -37,7 +37,9 @@ IF(LIBUSB_FOUND) ENDIF(MSVC) SET(HAVE_USB_SUPPORT TRUE) ELSE(LIBUSB_FOUND) - #TODO dummy usb + LIBUHD_APPEND_SOURCES( + ${CMAKE_SOURCE_DIR}/lib/transport/usb_dummy_impl.cpp + ) ENDIF(LIBUSB_FOUND) IF(HAVE_USB_SUPPORT) diff --git a/host/lib/transport/usb_dummy_impl.cpp b/host/lib/transport/usb_dummy_impl.cpp new file mode 100644 index 000000000..518342aba --- /dev/null +++ b/host/lib/transport/usb_dummy_impl.cpp @@ -0,0 +1,39 @@ +// +// 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 +#include +#include +#include + +using namespace uhd::transport; + +std::vector usb_device_handle::get_device_list(boost::uint16_t, boost::uint16_t){ + return std::vector(); //empty list +} + +usb_control::sptr usb_control::make(usb_device_handle::sptr){ + throw std::runtime_error("no usb support -> usb_control::make not implemented"); +} + +usb_zero_copy::sptr usb_zero_copy::make( + usb_device_handle::sptr, + unsigned int, unsigned int, + size_t, size_t, size_t, size_t +){ + throw std::runtime_error("no usb support -> usb_zero_copy::make not implemented"); +} -- 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') 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 d2494b0313399b141913ad332315fefbba012e94 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Tue, 5 Oct 2010 12:50:53 -0700 Subject: uhd: transport docs for UDP and USB (moved from usrp docs) --- host/docs/CMakeLists.txt | 1 + host/docs/index.rst | 9 ++-- host/docs/transport.rst | 81 +++++++++++++++++++++++++++++++ host/docs/usrp1.rst | 23 --------- host/docs/usrp2.rst | 41 ---------------- host/lib/transport/udp_zero_copy_asio.cpp | 5 +- 6 files changed, 91 insertions(+), 69 deletions(-) create mode 100644 host/docs/transport.rst (limited to 'host/lib/transport') diff --git a/host/docs/CMakeLists.txt b/host/docs/CMakeLists.txt index bbb8812b0..65db3befc 100644 --- a/host/docs/CMakeLists.txt +++ b/host/docs/CMakeLists.txt @@ -25,6 +25,7 @@ SET(manual_sources dboards.rst general.rst images.rst + transport.rst usrp1.rst usrp2.rst ) diff --git a/host/docs/index.rst b/host/docs/index.rst index bd55edc0b..7f8129e2d 100644 --- a/host/docs/index.rst +++ b/host/docs/index.rst @@ -20,11 +20,12 @@ Building the UHD ^^^^^^^^^^^^^^^^^^^^^ Application Notes ^^^^^^^^^^^^^^^^^^^^^ -* `General App Notes <./general.html>`_ +* `General Application Notes <./general.html>`_ * `Firmware and FPGA Image Notes <./images.html>`_ -* `USRP1 App Notes <./usrp1.html>`_ -* `USRP2 App Notes <./usrp2.html>`_ -* `Daughterboard App Notes <./dboards.html>`_ +* `USRP1 Application Notes <./usrp1.html>`_ +* `USRP2 Application Notes <./usrp2.html>`_ +* `Daughterboard Application Notes <./dboards.html>`_ +* `Transport Application Notes <./transport.html>`_ ^^^^^^^^^^^^^^^^^^^^^ API Documentation diff --git a/host/docs/transport.rst b/host/docs/transport.rst new file mode 100644 index 000000000..d6a146c67 --- /dev/null +++ b/host/docs/transport.rst @@ -0,0 +1,81 @@ +======================================================================== +UHD - Transport Application Notes +======================================================================== + +.. contents:: Table of Contents + +The advanced user can pass optional parameters +into the underlying transport layer through the device address. +These optional parameters control how the transport object allocates memory, +resizes kernel buffers, spawns threads, etc. +When not spcified, the transport layer will use values for these parameters +that are known to perform well on a variety of systems. +The transport parameters are defined below for the various transports in the UHD: + +------------------------------------------------------------------------ +UDP transport (ASIO) +------------------------------------------------------------------------ +The UDP transport is implemented with Boost's ASIO library. +ASIO provides an asynchronous API for user-space sockets. +The transport implementation allocates a number of buffers +and submits asynchronous requests for send and receive. +IO service threads run in the background to process these requests. + +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +Transport parameters +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +The following parameters can be used to alter the transport's default behavior: + +* **recv_frame_size:** The size of a single receive buffer in bytes +* **num_recv_frames:** The number of receive buffers to allocate +* **send_frame_size:** The size of a single send buffer in bytes +* **num_send_frames:** The number of send buffers to allocate +* **concurrency_hint:** The number of threads to run the IO service + +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +Resize socket buffers +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +It may be useful increase the size of the socket buffers to +move the burden of buffering samples into the kernel, or to +buffer incoming samples faster than they can be processed. +However, if your application cannot process samples fast enough, +no amount of buffering can save you. +The following parameters can be used to alter socket's buffer sizes: + +* **recv_buff_size:** The desired size of the receive buffer in bytes +* **send_buff_size:** The desired size of the send buffer in bytes + +**Note:** Large send buffers tend to decrease transmit performance. + +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +Linux specific notes +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +On linux, the maximum buffer sizes are capped by the sysctl values +**net.core.rmem_max** and **net.core.wmem_max**. +To change the maximum values, run the following commands: +:: + + sudo sysctl -w net.core.rmem_max= + sudo sysctl -w net.core.wmem_max= + +Set the values permanently by editing */etc/sysctl.conf* + +------------------------------------------------------------------------ +USB transport (libusb) +------------------------------------------------------------------------ +The USB transport is implemented with libusb. +Libusb provides an asynchronous API for USB bulk transfers. +The transport implementation allocates a number of buffers +and submits asynchronous requests through libusb. +A single thread runs in the background +and executes the libusb event handler to process these requests. + +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +Transport parameters +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +The following parameters can be used to alter the transport's default behavior: + +* **recv_frame_size:** The size of a single receive transfers in bytes +* **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 diff --git a/host/docs/usrp1.rst b/host/docs/usrp1.rst index 0baa93a45..3443fd871 100644 --- a/host/docs/usrp1.rst +++ b/host/docs/usrp1.rst @@ -60,29 +60,6 @@ Example device address string representations to specify non-standard firmware a fpga=usrp1_fpga_4rx.rbf, fw=usrp1_fw_custom.ihx -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -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 parameters 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 usage, set the device address markup string to the following: -:: - - serial=12345678, recv_num_xfers=16 - - -- OR -- - - serial=12345678, recv_xfer_size=2048, recv_num_xfers=16 - ------------------------------------------------------------------------ Specifying the subdevice to use ------------------------------------------------------------------------ diff --git a/host/docs/usrp2.rst b/host/docs/usrp2.rst index 70e5ea28b..1ebab388a 100644 --- a/host/docs/usrp2.rst +++ b/host/docs/usrp2.rst @@ -165,47 +165,6 @@ The device address string representation for 2 USRP2s with IPv4 addresses 192.16 addr=192.168.10.2 192.168.20.2 ------------------------------------------------------------------------- -Resize the send and receive buffers ------------------------------------------------------------------------- -It may be useful increase the size of the socket buffers to -move the burden of buffering samples into the kernel, or to -buffer incoming samples faster than they can be processed. -However, if you application cannot process samples fast enough, -no amount of buffering can save you. - -By default, the UHD will try to resize both the send and receive buffer for optimum performance. -A warning will be printed on instantiation if the actual buffer size is insufficient. -See the OS specific notes below: - -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -OS specific notes -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -On linux, the maximum buffer sizes are capped by the sysctl values -**net.core.rmem_max** and **net.core.wmem_max**. -To change the maximum values, run the following commands: -:: - - sudo sysctl -w net.core.rmem_max= - sudo sysctl -w net.core.wmem_max= - -Set the values permanently by editing */etc/sysctl.conf* - -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -Device address params -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -To manually set the size of the buffers, -the usrp2 will accept two optional parameters in the device address. -Each parameter will accept a numeric value for the number of bytes. - -* recv_buff_size -* send_buff_size - -Example usage, set the device address markup string to the following: -:: - - addr=192.168.10.2, recv_buff_size=100e6 - ------------------------------------------------------------------------ Hardware setup notes ------------------------------------------------------------------------ diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index ada282e07..a87f17a91 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -248,7 +248,10 @@ template static void resize_buff_helper( if (actual_size < target_size) uhd::print_warning(str(boost::format( "The %s buffer is smaller than the requested size.\n" "The minimum recommended buffer size is %d bytes.\n" - "See the USRP2 application notes on buffer resizing.\n" + "See the transport application notes on buffer resizing.\n" + #if defined(UHD_PLATFORM_LINUX) + "On Linux: sudo sysctl -w net.core.rmem_max=%2%\n" + #endif /*defined(UHD_PLATFORM_LINUX)*/ ) % name % min_sock_buff_size)); } -- 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') 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 From 7c127b6c82db773ca2dcab372ab28b4dc4e5b347 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Wed, 6 Oct 2010 10:34:10 -0700 Subject: udp: fixed boost format syntax for warning message --- host/lib/transport/udp_zero_copy_asio.cpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'host/lib/transport') diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index a87f17a91..7e28caf2d 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -246,11 +246,11 @@ template static void resize_buff_helper( "Current %s sock buff size: %d bytes" ) % name % actual_size << std::endl; if (actual_size < target_size) uhd::print_warning(str(boost::format( - "The %s buffer is smaller than the requested size.\n" - "The minimum recommended buffer size is %d bytes.\n" + "The %1% buffer is smaller than the requested size.\n" + "The minimum recommended buffer size is %2% bytes.\n" "See the transport application notes on buffer resizing.\n" #if defined(UHD_PLATFORM_LINUX) - "On Linux: sudo sysctl -w net.core.rmem_max=%2%\n" + "Please run: sudo sysctl -w net.core.rmem_max=%2%\n" #endif /*defined(UHD_PLATFORM_LINUX)*/ ) % name % min_sock_buff_size)); } -- cgit v1.2.3 From 952db7702f52e34fa05b8795abe46fb22add137d Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Thu, 7 Oct 2010 16:31:41 -0700 Subject: udp: implementation for blocking recv w/ timeout, switch async implementation w/ #define --- host/lib/transport/udp_zero_copy_asio.cpp | 140 +++++++++++++++++++++--------- 1 file changed, 101 insertions(+), 39 deletions(-) (limited to 'host/lib/transport') diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 7e28caf2d..798cc657d 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -35,6 +35,10 @@ namespace asio = boost::asio; /*********************************************************************** * Constants **********************************************************************/ +//Define this to the the boost async io calls to perform receive. +//Otherwise, get_recv_buff uses a blocking receive with timeout. +//#define USE_ASIO_ASYNC_RECV + //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); @@ -43,8 +47,15 @@ static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(4 * 25e6 * 0.5); //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_FRAMES = 32; +//The number of async frames to allocate for each send and recv: +//The non-async recv can have a very large number of recv frames +//because the CPU overhead is independent of the number of frames. +#ifdef USE_ASIO_ASYNC_RECV +static const size_t DEFAULT_NUM_RECV_FRAMES = 32; +#else +static const size_t DEFAULT_NUM_RECV_FRAMES = MIN_RECV_SOCK_BUFF_SIZE/udp_simple::mtu; +#endif +static const size_t DEFAULT_NUM_SEND_FRAMES = 32; //a single concurrent thread for io_service seems to be the fastest static const size_t CONCURRENCY_HINT = 1; @@ -67,9 +78,9 @@ public: ): _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))), + _num_recv_frames(size_t(hints.cast("num_recv_frames", DEFAULT_NUM_RECV_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))) + _num_send_frames(size_t(hints.cast("num_send_frames", DEFAULT_NUM_SEND_FRAMES))) { //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl; @@ -82,6 +93,13 @@ public: _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_asio_impl(void){ + delete _work; //allow io_service run to complete + _thread_group.join_all(); //wait for service threads to exit + delete _socket; } void init(void){ @@ -106,10 +124,9 @@ public: ); } - ~udp_zero_copy_asio_impl(void){ - delete _work; //allow io_service run to complete - _thread_group.join_all(); //wait for service threads to exit - delete _socket; + void service(void){ + set_thread_priority_safe(); + _io_service.run(); } //get size for internal socket buffer @@ -126,6 +143,9 @@ public: return get_buff_size(); } + //////////////////////////////////////////////////////////////////// + #ifdef USE_ASIO_ASYNC_RECV + //////////////////////////////////////////////////////////////////// //! 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 @@ -142,6 +162,74 @@ public: return managed_recv_buffer::sptr(); } + //! handle a recv callback -> push the filled memory into the fifo + void handle_recv(void *mem, size_t len){ + boost::this_thread::disable_interruption di; //disable because the wait can throw + _pending_recv_buffs->push_with_wait(boost::asio::buffer(mem, len)); + } + + //! release a recv buffer -> start an async recv on the buffer + void release(void *mem){ + _socket->async_receive( + boost::asio::buffer(mem, _recv_frame_size), + boost::bind( + &udp_zero_copy_asio_impl::handle_recv, + shared_from_this(), mem, + asio::placeholders::bytes_transferred + ) + ); + } + + //////////////////////////////////////////////////////////////////// + #else /*USE_ASIO_ASYNC_RECV*/ + //////////////////////////////////////////////////////////////////// + managed_recv_buffer::sptr get_recv_buff(double timeout){ + boost::this_thread::disable_interruption di; //disable because the wait can throw + + //setup timeval for timeout + timeval tv; + tv.tv_sec = 0; + tv.tv_usec = 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 managed_recv_buffer::sptr(); + + //grab an available buffer + asio::mutable_buffer buff; + if (not _pending_recv_buffs->pop_with_timed_wait(buff, timeout)) + return managed_recv_buffer::sptr(); + + //receive and return the buffer + return managed_recv_buffer::make_safe( + asio::buffer( + boost::asio::buffer_cast(buff), + _socket->receive(boost::asio::buffer(buff)) + ), + boost::bind( + &udp_zero_copy_asio_impl::release, + shared_from_this(), + asio::buffer_cast(buff) + ) + ); + } + + void release(void *mem){ + boost::this_thread::disable_interruption di; //disable because the wait can throw + _pending_recv_buffs->push_with_wait( + boost::asio::buffer(mem, this->get_recv_frame_size()) + ); + } + + //////////////////////////////////////////////////////////////////// + #endif /*USE_ASIO_ASYNC_RECV*/ + //////////////////////////////////////////////////////////////////// + size_t get_num_recv_frames(void) const {return _num_recv_frames;} size_t get_recv_frame_size(void) const {return _recv_frame_size;} @@ -161,37 +249,6 @@ public: return managed_send_buffer::sptr(); } - 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){ - set_thread_priority_safe(); - _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, _recv_frame_size), - boost::bind( - &udp_zero_copy_asio_impl::handle_recv, - shared_from_this(), mem, - asio::placeholders::bytes_transferred - ) - ); - } - //! 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 @@ -209,10 +266,15 @@ private: ); } + 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: //asio guts -> socket and service asio::ip::udp::socket *_socket; asio::io_service _io_service; asio::io_service::work *_work; + asio::ip::udp::socket::native_type _sock_fd; //memory management -> buffers and fifos boost::thread_group _thread_group; -- cgit v1.2.3 From 201af62ccb4d1065a44822f0f3ce4c81755510b5 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Thu, 7 Oct 2010 16:34:29 -0700 Subject: udp: fix msvc errors for udp transport --- host/lib/transport/udp_zero_copy_asio.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'host/lib/transport') diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 798cc657d..40fe3fa0f 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -189,7 +189,7 @@ public: //setup timeval for timeout timeval tv; tv.tv_sec = 0; - tv.tv_usec = timeout*1e6; + tv.tv_usec = long(timeout*1e6); //setup rset for timeout fd_set rset; @@ -274,7 +274,7 @@ private: asio::ip::udp::socket *_socket; asio::io_service _io_service; asio::io_service::work *_work; - asio::ip::udp::socket::native_type _sock_fd; + int _sock_fd; //memory management -> buffers and fifos boost::thread_group _thread_group; -- cgit v1.2.3 From e8e9258acb8ce8499c40de52abaa4e6ba83d479d Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Fri, 8 Oct 2010 11:20:30 -0700 Subject: udp: worked blocking send back into udp transport, enable async with #define tweaked some code in bounded buffer --- host/include/uhd/transport/alignment_buffer.ipp | 8 +- host/include/uhd/transport/bounded_buffer.ipp | 39 +++++--- host/lib/transport/udp_zero_copy_asio.cpp | 114 +++++++++++++++--------- 3 files changed, 99 insertions(+), 62 deletions(-) (limited to 'host/lib/transport') diff --git a/host/include/uhd/transport/alignment_buffer.ipp b/host/include/uhd/transport/alignment_buffer.ipp index 5f09de0d9..833b5d399 100644 --- a/host/include/uhd/transport/alignment_buffer.ipp +++ b/host/include/uhd/transport/alignment_buffer.ipp @@ -54,14 +54,14 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE bool pop_elems_with_timed_wait( std::vector &elems, double timeout ){ - boost::system_time exit_time = boost::get_system_time() + boost::posix_time::microseconds(long(timeout*1e6)); + boost::system_time exit_time = boost::get_system_time() + to_time_dur(timeout); buff_contents_type buff_contents_tmp; std::list indexes_to_do(_all_indexes); //do an initial pop to load an initial sequence id size_t index = indexes_to_do.front(); if (not _buffs[index]->pop_with_timed_wait( - buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() + buff_contents_tmp, from_time_dur(exit_time - boost::get_system_time()) )) return false; elems[index] = buff_contents_tmp.first; seq_type expected_seq_id = buff_contents_tmp.second; @@ -76,7 +76,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ indexes_to_do = _all_indexes; index = indexes_to_do.front(); if (not _buffs[index]->pop_with_timed_wait( - buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() + buff_contents_tmp, from_time_dur(exit_time - boost::get_system_time()) )) return false; elems[index] = buff_contents_tmp.first; expected_seq_id = buff_contents_tmp.second; @@ -86,7 +86,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ //pop an element off for this index index = indexes_to_do.front(); if (not _buffs[index]->pop_with_timed_wait( - buff_contents_tmp, 1e-6*(exit_time - boost::get_system_time()).total_microseconds() + buff_contents_tmp, from_time_dur(exit_time - boost::get_system_time()) )) return false; //if the sequence id matches: diff --git a/host/include/uhd/transport/bounded_buffer.ipp b/host/include/uhd/transport/bounded_buffer.ipp index 58f78bab4..edc7faa06 100644 --- a/host/include/uhd/transport/bounded_buffer.ipp +++ b/host/include/uhd/transport/bounded_buffer.ipp @@ -19,18 +19,28 @@ #define INCLUDED_UHD_TRANSPORT_BOUNDED_BUFFER_IPP #include +#include #include #include #include namespace uhd{ namespace transport{ namespace{ /*anon*/ + static UHD_INLINE boost::posix_time::time_duration to_time_dur(double timeout){ + return boost::posix_time::microseconds(long(timeout*1e6)); + } + + static UHD_INLINE double from_time_dur(const boost::posix_time::time_duration &time_dur){ + return 1e-6*time_dur.total_microseconds(); + } + template class bounded_buffer_impl : public bounded_buffer{ public: bounded_buffer_impl(size_t capacity) : _buffer(capacity){ - /* NOP */ + _not_full_fcn = boost::bind(&bounded_buffer_impl::not_full, this); + _not_empty_fcn = boost::bind(&bounded_buffer_impl::not_empty, this); } UHD_INLINE bool push_with_pop_on_full(const elem_type &elem){ @@ -52,17 +62,16 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE void push_with_wait(const elem_type &elem){ boost::unique_lock lock(_mutex); - _full_cond.wait(lock, boost::bind(&bounded_buffer_impl::not_full, this)); + _full_cond.wait(lock, _not_full_fcn); _buffer.push_front(elem); lock.unlock(); _empty_cond.notify_one(); } - bool push_with_timed_wait(const elem_type &elem, double timeout){ + UHD_INLINE bool push_with_timed_wait(const elem_type &elem, double timeout){ boost::unique_lock lock(_mutex); if (not _full_cond.timed_wait( - lock, boost::posix_time::microseconds(long(timeout*1e6)), - boost::bind(&bounded_buffer_impl::not_full, this) + lock, to_time_dur(timeout), _not_full_fcn )) return false; _buffer.push_front(elem); lock.unlock(); @@ -72,19 +81,18 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE void pop_with_wait(elem_type &elem){ boost::unique_lock lock(_mutex); - _empty_cond.wait(lock, boost::bind(&bounded_buffer_impl::not_empty, this)); - this->pop_back(elem); + _empty_cond.wait(lock, _not_empty_fcn); + elem = this->pop_back(); lock.unlock(); _full_cond.notify_one(); } - bool pop_with_timed_wait(elem_type &elem, double timeout){ + UHD_INLINE bool pop_with_timed_wait(elem_type &elem, double timeout){ boost::unique_lock lock(_mutex); if (not _empty_cond.timed_wait( - lock, boost::posix_time::microseconds(long(timeout*1e6)), - boost::bind(&bounded_buffer_impl::not_empty, this) + lock, to_time_dur(timeout), _not_empty_fcn )) return false; - this->pop_back(elem); + elem = this->pop_back(); lock.unlock(); _full_cond.notify_one(); return true; @@ -92,7 +100,7 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ UHD_INLINE void clear(void){ boost::unique_lock lock(_mutex); - while (not_empty()) _buffer.pop_back(); + while (not_empty()) this->pop_back(); lock.unlock(); _full_cond.notify_one(); } @@ -105,16 +113,19 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/ bool not_full(void) const{return not _buffer.full();} bool not_empty(void) const{return not _buffer.empty();} + boost::function _not_full_fcn, _not_empty_fcn; + /*! * Three part operation to pop an element: * 1) assign elem to the back element * 2) assign the back element to empty * 3) pop the back to move the counter */ - UHD_INLINE void pop_back(elem_type &elem){ - elem = _buffer.back(); + UHD_INLINE elem_type pop_back(void){ + elem_type elem = _buffer.back(); _buffer.back() = elem_type(); _buffer.pop_back(); + return elem; } }; }}} //namespace diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index 40fe3fa0f..d84aeefdd 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -39,6 +39,10 @@ namespace asio = boost::asio; //Otherwise, get_recv_buff uses a blocking receive with timeout. //#define USE_ASIO_ASYNC_RECV +//Define this to the the boost async io calls to perform send. +//Otherwise, the commit callback uses a blocking send. +//#define USE_ASIO_ASYNC_SEND + //enough buffering for half a second of samples at full rate on usrp2 static const size_t MIN_RECV_SOCK_BUFF_SIZE = size_t(4 * 25e6 * 0.5); @@ -55,7 +59,13 @@ static const size_t DEFAULT_NUM_RECV_FRAMES = 32; #else static const size_t DEFAULT_NUM_RECV_FRAMES = MIN_RECV_SOCK_BUFF_SIZE/udp_simple::mtu; #endif +//The non-async send only ever requires a single frame +//because the buffer will be committed before a new get. +#ifdef USE_ASIO_ASYNC_SEND static const size_t DEFAULT_NUM_SEND_FRAMES = 32; +#else +static const size_t DEFAULT_NUM_SEND_FRAMES = MIN_SEND_SOCK_BUFF_SIZE/udp_simple::mtu;; +#endif //a single concurrent thread for io_service seems to be the fastest static const size_t CONCURRENCY_HINT = 1; @@ -143,6 +153,12 @@ public: return get_buff_size(); } + //! handle a recv callback -> push the filled memory into the fifo + UHD_INLINE void handle_recv(void *mem, size_t len){ + boost::this_thread::disable_interruption di; //disable because the wait can throw + _pending_recv_buffs->push_with_wait(boost::asio::buffer(mem, len)); + } + //////////////////////////////////////////////////////////////////// #ifdef USE_ASIO_ASYNC_RECV //////////////////////////////////////////////////////////////////// @@ -162,16 +178,10 @@ public: return managed_recv_buffer::sptr(); } - //! handle a recv callback -> push the filled memory into the fifo - void handle_recv(void *mem, size_t len){ - boost::this_thread::disable_interruption di; //disable because the wait can throw - _pending_recv_buffs->push_with_wait(boost::asio::buffer(mem, len)); - } - //! release a recv buffer -> start an async recv on the buffer void release(void *mem){ _socket->async_receive( - boost::asio::buffer(mem, _recv_frame_size), + boost::asio::buffer(mem, this->get_recv_frame_size()), boost::bind( &udp_zero_copy_asio_impl::handle_recv, shared_from_this(), mem, @@ -185,6 +195,7 @@ public: //////////////////////////////////////////////////////////////////// managed_recv_buffer::sptr get_recv_buff(double timeout){ boost::this_thread::disable_interruption di; //disable because the wait can throw + asio::mutable_buffer buff; //setup timeval for timeout timeval tv; @@ -196,34 +207,30 @@ public: FD_ZERO(&rset); FD_SET(_sock_fd, &rset); - //call select to perform timed wait - if (::select(_sock_fd+1, &rset, NULL, NULL, &tv) <= 0) - return managed_recv_buffer::sptr(); - - //grab an available buffer - asio::mutable_buffer buff; - if (not _pending_recv_buffs->pop_with_timed_wait(buff, timeout)) - return managed_recv_buffer::sptr(); - - //receive and return the buffer - return managed_recv_buffer::make_safe( - asio::buffer( - boost::asio::buffer_cast(buff), - _socket->receive(boost::asio::buffer(buff)) - ), - boost::bind( - &udp_zero_copy_asio_impl::release, - shared_from_this(), - asio::buffer_cast(buff) - ) - ); + //call select to perform timed wait and grab an available buffer with wait + //if the condition is true, call receive and return the managed buffer + if ( + ::select(_sock_fd+1, &rset, NULL, NULL, &tv) > 0 and + _pending_recv_buffs->pop_with_timed_wait(buff, timeout) + ){ + return managed_recv_buffer::make_safe( + asio::buffer( + boost::asio::buffer_cast(buff), + _socket->receive(asio::buffer(buff)) + ), + boost::bind( + &udp_zero_copy_asio_impl::release, + shared_from_this(), + asio::buffer_cast(buff) + ) + ); + } + return managed_recv_buffer::sptr(); } void release(void *mem){ boost::this_thread::disable_interruption di; //disable because the wait can throw - _pending_recv_buffs->push_with_wait( - boost::asio::buffer(mem, this->get_recv_frame_size()) - ); + handle_recv(mem, this->get_recv_frame_size()); } //////////////////////////////////////////////////////////////////// @@ -233,6 +240,12 @@ public: size_t get_num_recv_frames(void) const {return _num_recv_frames;} size_t get_recv_frame_size(void) const {return _recv_frame_size;} + //! handle a send callback -> push the emptied memory into the fifo + UHD_INLINE void handle_send(void *mem){ + boost::this_thread::disable_interruption di; //disable because the wait can throw + _pending_send_buffs->push_with_wait(boost::asio::buffer(mem, this->get_send_frame_size())); + } + //! pop an empty send buffer off of the fifo and bind with the commit callback managed_send_buffer::sptr get_send_buff(double timeout){ boost::this_thread::disable_interruption di; //disable because the wait can throw @@ -249,12 +262,9 @@ public: return managed_send_buffer::sptr(); } - //! handle a send callback -> push the emptied memory into the fifo - void handle_send(void *mem){ - boost::this_thread::disable_interruption di; //disable because the wait can throw - _pending_send_buffs->push_with_wait(boost::asio::buffer(mem, _send_frame_size)); - } - + //////////////////////////////////////////////////////////////////// + #ifdef USE_ASIO_ASYNC_SEND + //////////////////////////////////////////////////////////////////// //! commit a send buffer -> start an async send on the buffer void commit(void *mem, size_t len){ _socket->async_send( @@ -266,6 +276,18 @@ public: ); } + //////////////////////////////////////////////////////////////////// + #else /*USE_ASIO_ASYNC_SEND*/ + //////////////////////////////////////////////////////////////////// + void commit(void *mem, size_t len){ + _socket->send(asio::buffer(mem, len)); + handle_send(mem); + } + + //////////////////////////////////////////////////////////////////// + #endif /*USE_ASIO_ASYNC_SEND*/ + //////////////////////////////////////////////////////////////////// + size_t get_num_send_frames(void) const {return _num_send_frames;} size_t get_send_frame_size(void) const {return _send_frame_size;} @@ -297,6 +319,13 @@ template static void resize_buff_helper( if (name == "recv") min_sock_buff_size = MIN_RECV_SOCK_BUFF_SIZE; if (name == "send") min_sock_buff_size = MIN_SEND_SOCK_BUFF_SIZE; + std::string help_message; + #if defined(UHD_PLATFORM_LINUX) + help_message = str(boost::format( + "Please run: sudo sysctl -w net.core.%smem_max=%d\n" + ) % ((name == "recv")?"r":"w") % min_sock_buff_size); + #endif /*defined(UHD_PLATFORM_LINUX)*/ + //resize the buffer if size was provided if (target_size > 0){ size_t actual_size = udp_trans->resize_buff(target_size); @@ -308,13 +337,10 @@ template static void resize_buff_helper( "Current %s sock buff size: %d bytes" ) % name % actual_size << std::endl; if (actual_size < target_size) uhd::print_warning(str(boost::format( - "The %1% buffer is smaller than the requested size.\n" - "The minimum recommended buffer size is %2% bytes.\n" - "See the transport application notes on buffer resizing.\n" - #if defined(UHD_PLATFORM_LINUX) - "Please run: sudo sysctl -w net.core.rmem_max=%2%\n" - #endif /*defined(UHD_PLATFORM_LINUX)*/ - ) % name % min_sock_buff_size)); + "The %s buffer is smaller than the requested size.\n" + "The minimum recommended buffer size is %d bytes.\n" + "See the transport application notes on buffer resizing.\n%s" + ) % name % min_sock_buff_size % help_message)); } //only enable on platforms that are happy with the large buffer resize -- cgit v1.2.3 From cea3d05095202413be12cd4792ab9f781cbccef7 Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Tue, 26 Oct 2010 14:14:01 -0700 Subject: uhd: replaced print warning with a post warning call and registry renamed print warning calls in the implementation fixed issue with dict::pop so it now works even if the value is not comparable --- host/include/uhd/types/dict.hpp | 23 ++++++++---- host/include/uhd/types/tune_request.hpp | 2 +- host/include/uhd/usrp/mimo_usrp.hpp | 4 +-- host/include/uhd/usrp/simple_usrp.hpp | 2 +- host/include/uhd/utils/warning.hpp | 36 ++++++++++++++++--- host/lib/transport/udp_zero_copy_asio.cpp | 2 +- host/lib/usrp/dboard/db_basic_and_lf.cpp | 4 +-- host/lib/usrp/dboard/db_dbsrx.cpp | 8 ++--- host/lib/usrp/dboard/db_rfx.cpp | 4 +-- host/lib/usrp/dboard/db_tvrx.cpp | 2 +- host/lib/usrp/dboard/db_unknown.cpp | 4 +-- host/lib/usrp/dboard/db_wbx.cpp | 4 +-- host/lib/usrp/dboard_manager.cpp | 4 +-- host/lib/usrp/multi_usrp.cpp | 2 +- host/lib/usrp/usrp1/mboard_impl.cpp | 2 +- host/lib/usrp/usrp1/usrp1_impl.cpp | 2 +- host/lib/usrp/wrapper_utils.hpp | 4 +-- host/lib/utils/thread_priority.cpp | 2 +- host/lib/utils/warning.cpp | 59 ++++++++++++++++++++++++++++--- host/test/warning_test.cpp | 4 +-- 20 files changed, 131 insertions(+), 43 deletions(-) (limited to 'host/lib/transport') diff --git a/host/include/uhd/types/dict.hpp b/host/include/uhd/types/dict.hpp index de96ea768..3d0acf888 100644 --- a/host/include/uhd/types/dict.hpp +++ b/host/include/uhd/types/dict.hpp @@ -120,10 +120,7 @@ namespace uhd{ BOOST_FOREACH(const pair_t &p, _map){ if (p.first == key) return p.second; } - throw std::invalid_argument(str(boost::format( - "key \"%s\" not found in dict(%s, %s)" - ) % boost::lexical_cast(key) - % typeid(Key).name() % typeid(Val).name())); + throw key_not_found_in_dict(key); } /*! @@ -147,12 +144,24 @@ namespace uhd{ * \throw an exception when not found */ Val pop(const Key &key){ - Val val = (*this)[key]; - _map.remove(pair_t(key, val)); - return val; + typename std::list::iterator it; + for (it = _map.begin(); it != _map.end(); it++){ + if (it->first != key) continue; + Val val = it->second; + _map.erase(it); + return val; + } + throw key_not_found_in_dict(key); } private: + std::exception key_not_found_in_dict(const Key &key) const{ + return std::out_of_range(str(boost::format( + "key \"%s\" not found in dict(%s, %s)" + ) % boost::lexical_cast(key) + % typeid(Key).name() % typeid(Val).name())); + } + std::list _map; //private container }; diff --git a/host/include/uhd/types/tune_request.hpp b/host/include/uhd/types/tune_request.hpp index b05ab5cc7..942b93251 100644 --- a/host/include/uhd/types/tune_request.hpp +++ b/host/include/uhd/types/tune_request.hpp @@ -60,7 +60,7 @@ namespace uhd{ /*! * The target frequency of the overall chain in Hz. - * Use in conjunction with the automatic policies. + * Set this even if all policies are set to manual. */ double target_freq; diff --git a/host/include/uhd/usrp/mimo_usrp.hpp b/host/include/uhd/usrp/mimo_usrp.hpp index b5acf84e1..a2092f04f 100644 --- a/host/include/uhd/usrp/mimo_usrp.hpp +++ b/host/include/uhd/usrp/mimo_usrp.hpp @@ -298,7 +298,7 @@ public: time_spec_t time_0 = _mboard(0)[MBOARD_PROP_TIME_NOW].as(); time_spec_t time_i = _mboard(chan)[MBOARD_PROP_TIME_NOW].as(); if (time_i < time_0 or (time_i - time_0) > time_spec_t(0.01)){ //10 ms: greater than RTT but not too big - uhd::print_warning(str(boost::format( + uhd::warning::post(str(boost::format( "Detected time deviation between board %d and board 0.\n" "Board 0 time is %f seconds.\n" "Board %d time is %f seconds.\n" @@ -512,7 +512,7 @@ namespace uhd{ namespace usrp{ * The Make Function **********************************************************************/ inline mimo_usrp::sptr mimo_usrp::make(const device_addr_t &dev_addr){ - uhd::print_warning( + uhd::warning::post( "The mimo USRP interface has been deprecated.\n" "Please switch to the multi USRP interface.\n" "#include \n" diff --git a/host/include/uhd/usrp/simple_usrp.hpp b/host/include/uhd/usrp/simple_usrp.hpp index 59fd9bb09..77416dbbd 100644 --- a/host/include/uhd/usrp/simple_usrp.hpp +++ b/host/include/uhd/usrp/simple_usrp.hpp @@ -374,7 +374,7 @@ namespace uhd{ namespace usrp{ * The Make Function **********************************************************************/ inline simple_usrp::sptr simple_usrp::make(const device_addr_t &dev_addr){ - uhd::print_warning( + uhd::warning::post( "The simple USRP interface has been deprecated.\n" "Please switch to the single USRP interface.\n" "#include \n" diff --git a/host/include/uhd/utils/warning.hpp b/host/include/uhd/utils/warning.hpp index 91d8400ab..a1e3f0d1e 100644 --- a/host/include/uhd/utils/warning.hpp +++ b/host/include/uhd/utils/warning.hpp @@ -19,16 +19,44 @@ #define INCLUDED_UHD_UTILS_WARNING_HPP #include +#include +#include #include -namespace uhd{ +namespace uhd{ namespace warning{ + + //! Callback function type for a message handler + typedef boost::function handler_t; /*! - * Print a formatted warning string to stderr. + * Post a warning message to all registered handlers. * \param msg the multiline warning message */ - UHD_API void print_warning(const std::string &msg); + UHD_API void post(const std::string &msg); + + /*! + * Register a new handler with this name. + * If the name was already registered for this name, + * the old registered handler will be replaced. + * \param name a unique name for this handler + * \param handler the callback handler function + */ + UHD_API void register_handler(const std::string &name, const handler_t &handler); + + /*! + * Unregister a handler for this name. + * \param name a unique name for a registered handler + * \return the handler that was registered + * \throw error when the name was not found in the registry + */ + UHD_API handler_t unregister_handler(const std::string &name); + + /*! + * Get a list of registered handler names. + * \return a vector of unique string names + */ + UHD_API const std::vector registry_names(void); -} //namespace uhd +}} //namespace uhd::warning #endif /* INCLUDED_UHD_UTILS_WARNING_HPP */ diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index d84aeefdd..ed29864e9 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -336,7 +336,7 @@ template static void resize_buff_helper( else std::cout << boost::format( "Current %s sock buff size: %d bytes" ) % name % actual_size << std::endl; - if (actual_size < target_size) uhd::print_warning(str(boost::format( + if (actual_size < target_size) uhd::warning::post(str(boost::format( "The %s buffer is smaller than the requested size.\n" "The minimum recommended buffer size is %d bytes.\n" "See the transport application notes on buffer resizing.\n%s" diff --git a/host/lib/usrp/dboard/db_basic_and_lf.cpp b/host/lib/usrp/dboard/db_basic_and_lf.cpp index 5a87f4ecf..4c49b3bff 100644 --- a/host/lib/usrp/dboard/db_basic_and_lf.cpp +++ b/host/lib/usrp/dboard/db_basic_and_lf.cpp @@ -191,7 +191,7 @@ void basic_rx::rx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("%s: No tunable bandwidth, fixed filtered to %0.2fMHz") % get_rx_id().to_pp_string() % _max_freq ) @@ -302,7 +302,7 @@ void basic_tx::tx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("%s: No tunable bandwidth, fixed filtered to %0.2fMHz") % get_tx_id().to_pp_string() % _max_freq ) diff --git a/host/lib/usrp/dboard/db_dbsrx.cpp b/host/lib/usrp/dboard/db_dbsrx.cpp index 10e7f292b..85251bdf9 100644 --- a/host/lib/usrp/dboard/db_dbsrx.cpp +++ b/host/lib/usrp/dboard/db_dbsrx.cpp @@ -175,7 +175,7 @@ UHD_STATIC_BLOCK(reg_dbsrx_dboard){ dbsrx::dbsrx(ctor_args_t args) : rx_dboard_base(args){ //warn user about incorrect DBID on USRP1, requires R193 populated if (this->get_iface()->get_special_props().soft_clock_divider and this->get_rx_id() == 0x000D) - uhd::print_warning( + uhd::warning::post( str(boost::format( "DBSRX: incorrect dbid\n" "Expected dbid 0x0002 and R193\n" @@ -186,7 +186,7 @@ dbsrx::dbsrx(ctor_args_t args) : rx_dboard_base(args){ //warn user about incorrect DBID on non-USRP1, requires R194 populated if (not this->get_iface()->get_special_props().soft_clock_divider and this->get_rx_id() == 0x0002) - uhd::print_warning( + uhd::warning::post( str(boost::format( "DBSRX: incorrect dbid\n" "Expected dbid 0x000D and R194\n" @@ -342,7 +342,7 @@ void dbsrx::set_lo_freq(double target_freq){ //vtune is too low, try lower frequency vco if (_max2118_read_regs.adc == 0){ if (_max2118_write_regs.osc_band == 0){ - uhd::print_warning( + uhd::warning::post( str(boost::format( "DBSRX: Tuning exceeded vco range, _max2118_write_regs.osc_band == %d\n" ) % int(_max2118_write_regs.osc_band)) @@ -356,7 +356,7 @@ void dbsrx::set_lo_freq(double target_freq){ //vtune is too high, try higher frequency vco if (_max2118_read_regs.adc == 7){ if (_max2118_write_regs.osc_band == 7){ - uhd::print_warning( + uhd::warning::post( str(boost::format( "DBSRX: Tuning exceeded vco range, _max2118_write_regs.osc_band == %d\n" ) % int(_max2118_write_regs.osc_band)) diff --git a/host/lib/usrp/dboard/db_rfx.cpp b/host/lib/usrp/dboard/db_rfx.cpp index 0eeb0bfdf..12e458d8c 100644 --- a/host/lib/usrp/dboard/db_rfx.cpp +++ b/host/lib/usrp/dboard/db_rfx.cpp @@ -487,7 +487,7 @@ void rfx_xcvr::rx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("RFX: No tunable bandwidth, fixed filtered to 40MHz")) ); return; @@ -584,7 +584,7 @@ void rfx_xcvr::tx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("RFX: No tunable bandwidth, fixed filtered to 40MHz")) ); return; diff --git a/host/lib/usrp/dboard/db_tvrx.cpp b/host/lib/usrp/dboard/db_tvrx.cpp index c7aecf195..1f3c76556 100644 --- a/host/lib/usrp/dboard/db_tvrx.cpp +++ b/host/lib/usrp/dboard/db_tvrx.cpp @@ -484,7 +484,7 @@ void tvrx::rx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("TVRX: No tunable bandwidth, fixed filtered to 6MHz")) ); return; diff --git a/host/lib/usrp/dboard/db_unknown.cpp b/host/lib/usrp/dboard/db_unknown.cpp index 11293a1ba..a342471c4 100644 --- a/host/lib/usrp/dboard/db_unknown.cpp +++ b/host/lib/usrp/dboard/db_unknown.cpp @@ -164,7 +164,7 @@ void unknown_rx::rx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("Unknown Daughterboard: No tunable bandwidth, fixed filtered to 0.0MHz")) ); return; @@ -270,7 +270,7 @@ void unknown_tx::tx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("Unknown Daughterboard: No tunable bandwidth, fixed filtered to 0.0MHz")) ); return; diff --git a/host/lib/usrp/dboard/db_wbx.cpp b/host/lib/usrp/dboard/db_wbx.cpp index b07f179a3..647f1b975 100644 --- a/host/lib/usrp/dboard/db_wbx.cpp +++ b/host/lib/usrp/dboard/db_wbx.cpp @@ -556,7 +556,7 @@ void wbx_xcvr::rx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("WBX: No tunable bandwidth, fixed filtered to 40MHz")) ); return; @@ -657,7 +657,7 @@ void wbx_xcvr::tx_set(const wax::obj &key_, const wax::obj &val){ return; //always enabled case SUBDEV_PROP_BANDWIDTH: - uhd::print_warning( + uhd::warning::post( str(boost::format("WBX: No tunable bandwidth, fixed filtered to 40MHz")) ); return; diff --git a/host/lib/usrp/dboard_manager.cpp b/host/lib/usrp/dboard_manager.cpp index d73a698ae..5a98bb8eb 100644 --- a/host/lib/usrp/dboard_manager.cpp +++ b/host/lib/usrp/dboard_manager.cpp @@ -189,7 +189,7 @@ static args_t get_dboard_args( //verify that there is a registered constructor for this id if (not get_id_to_args_map().has_key(dboard_id)){ - uhd::print_warning(str(boost::format( + uhd::warning::post(str(boost::format( "Unknown dboard ID: %s.\n" ) % dboard_id.to_pp_string())); return get_dboard_args(unit, dboard_id, true); @@ -217,7 +217,7 @@ dboard_manager_impl::dboard_manager_impl( //warn for invalid dboard id xcvr combinations if (rx_dboard_is_xcvr != this_dboard_is_xcvr or tx_dboard_is_xcvr != this_dboard_is_xcvr){ - uhd::print_warning(str(boost::format( + uhd::warning::post(str(boost::format( "Unknown transceiver board ID combination...\n" "RX dboard ID: %s\n" "TX dboard ID: %s\n" diff --git a/host/lib/usrp/multi_usrp.cpp b/host/lib/usrp/multi_usrp.cpp index 8b82502ba..024a6152d 100644 --- a/host/lib/usrp/multi_usrp.cpp +++ b/host/lib/usrp/multi_usrp.cpp @@ -145,7 +145,7 @@ public: time_spec_t time_0 = _mboard(0)[MBOARD_PROP_TIME_NOW].as(); time_spec_t time_i = _mboard(m)[MBOARD_PROP_TIME_NOW].as(); if (time_i < time_0 or (time_i - time_0) > time_spec_t(0.01)){ //10 ms: greater than RTT but not too big - uhd::print_warning(str(boost::format( + uhd::warning::post(str(boost::format( "Detected time deviation between board %d and board 0.\n" "Board 0 time is %f seconds.\n" "Board %d time is %f seconds.\n" diff --git a/host/lib/usrp/usrp1/mboard_impl.cpp b/host/lib/usrp/usrp1/mboard_impl.cpp index fe3774eb4..669b20efa 100644 --- a/host/lib/usrp/usrp1/mboard_impl.cpp +++ b/host/lib/usrp/usrp1/mboard_impl.cpp @@ -98,7 +98,7 @@ static boost::uint32_t calc_rx_mux( // for all quadrature sources: Z = 0 // for mixed sources: warning + Z = 0 int Z = (num_quads > 0)? 0 : 1; - if (num_quads != 0 and num_reals != 0) uhd::print_warning( + if (num_quads != 0 and num_reals != 0) uhd::warning::post( "Mixing real and quadrature rx subdevices is not supported.\n" "The Q input to the real source(s) will be non-zero.\n" ); diff --git a/host/lib/usrp/usrp1/usrp1_impl.cpp b/host/lib/usrp/usrp1/usrp1_impl.cpp index 276ca86f6..314384e72 100644 --- a/host/lib/usrp/usrp1/usrp1_impl.cpp +++ b/host/lib/usrp/usrp1/usrp1_impl.cpp @@ -64,7 +64,7 @@ static device_addrs_t usrp1_find(const device_addr_t &hint) ); } catch(...){ - uhd::print_warning( + uhd::warning::post( "Could not locate USRP1 firmware.\n" "Please install the images package.\n" ); diff --git a/host/lib/usrp/wrapper_utils.hpp b/host/lib/usrp/wrapper_utils.hpp index aee230fc0..6f9fdbfca 100644 --- a/host/lib/usrp/wrapper_utils.hpp +++ b/host/lib/usrp/wrapper_utils.hpp @@ -40,7 +40,7 @@ static inline void do_samp_rate_warning_message( ){ static const double max_allowed_error = 1.0; //Sps if (std::abs(target_rate - actual_rate) > max_allowed_error){ - uhd::print_warning(str(boost::format( + uhd::warning::post(str(boost::format( "The hardware does not support the requested %s sample rate:\n" "Target sample rate: %f MSps\n" "Actual sample rate: %f MSps\n" @@ -55,7 +55,7 @@ static inline void do_tune_freq_warning_message( ){ static const double max_allowed_error = 1.0; //Hz if (std::abs(target_freq - actual_freq) > max_allowed_error){ - uhd::print_warning(str(boost::format( + uhd::warning::post(str(boost::format( "The hardware does not support the requested %s frequency:\n" "Target frequency: %f MHz\n" "Actual frequency: %f MHz\n" diff --git a/host/lib/utils/thread_priority.cpp b/host/lib/utils/thread_priority.cpp index f09d1b1d6..40b74f655 100644 --- a/host/lib/utils/thread_priority.cpp +++ b/host/lib/utils/thread_priority.cpp @@ -26,7 +26,7 @@ bool uhd::set_thread_priority_safe(float priority, bool realtime){ set_thread_priority(priority, realtime); return true; }catch(const std::exception &e){ - uhd::print_warning(str(boost::format( + uhd::warning::post(str(boost::format( "%s\n" "Failed to set thread priority %d (%s):\n" "Performance may be negatively affected.\n" diff --git a/host/lib/utils/warning.cpp b/host/lib/utils/warning.cpp index 8a7d35a23..05be7ae4d 100644 --- a/host/lib/utils/warning.cpp +++ b/host/lib/utils/warning.cpp @@ -17,16 +17,67 @@ #include #include +#include +#include #include +#include +#include #include #include using namespace uhd; -void uhd::print_warning(const std::string &msg){ - //print the warning message - std::cerr << std::endl << "Warning:" << std::endl; +/*********************************************************************** + * Registry implementation + **********************************************************************/ +//create the registry for the handlers +typedef uhd::dict registry_t; +UHD_SINGLETON_FCN(registry_t, get_registry) + +//the default warning handler +static void stderr_warning(const std::string &msg){ + std::cerr << msg; +} + +//register a default handler +UHD_STATIC_BLOCK(warning_register_default){ + warning::register_handler("default", &stderr_warning); +} + +/*********************************************************************** + * Post + format + **********************************************************************/ +void warning::post(const std::string &msg){ + std::stringstream ss; + + //format the warning message + ss << std::endl << "Warning:" << std::endl; BOOST_FOREACH(const std::string &line, std::split_string(msg, "\n")){ - std::cerr << " " << line << std::endl; + ss << " " << line << std::endl; + } + + //post the formatted message + BOOST_FOREACH(const std::string &name, get_registry().keys()){ + get_registry()[name](ss.str()); } } + +/*********************************************************************** + * Registry accessor functions + **********************************************************************/ +void warning::register_handler( + const std::string &name, const handler_t &handler +){ + get_registry()[name] = handler; +} + +warning::handler_t warning::unregister_handler(const std::string &name){ + if (not get_registry().has_key(name)) throw std::runtime_error( + "The warning registry does not have a handler registered to " + name + ); + return get_registry().pop(name); +} + +const std::vector warning::registry_names(void){ + return get_registry().keys(); +} diff --git a/host/test/warning_test.cpp b/host/test/warning_test.cpp index 6202c4270..db19955de 100644 --- a/host/test/warning_test.cpp +++ b/host/test/warning_test.cpp @@ -19,9 +19,9 @@ #include #include -BOOST_AUTO_TEST_CASE(test_print_warning){ +BOOST_AUTO_TEST_CASE(test_warning_post){ std::cerr << "---begin print test ---" << std::endl; - uhd::print_warning( + uhd::warning::post( "This is a test print for a warning message.\n" "And this is the second line of the test print.\n" ); -- cgit v1.2.3