path: root/host/lib/transport/libusb1_zero_copy.cpp

//
// Copyright 2010-2013 Ettus Research LLC
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
//

#include "libusb1_base.hpp"
#include <uhd/transport/usb_zero_copy.hpp>
#include <uhd/transport/buffer_pool.hpp>
#include <uhd/transport/bounded_buffer.hpp>
#include <uhd/utils/msg.hpp>
#include <uhd/exception.hpp>
#include <boost/foreach.hpp>
#include <boost/format.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <boost/make_shared.hpp>
#include <boost/circular_buffer.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>
#include <list>

using namespace uhd;
using namespace uhd::transport;

static const size_t DEFAULT_NUM_XFERS = 16;     //num xfers
static const size_t DEFAULT_XFER_SIZE = 32*512; //bytes

//! Define LIBUSB_CALL when its missing (non-windows)
#ifndef LIBUSB_CALL
    #define LIBUSB_CALL
#endif /*LIBUSB_CALL*/

//! libusb_handle_events_timeout_completed is only in newer API
#ifndef HAVE_LIBUSB_HANDLE_EVENTS_TIMEOUT_COMPLETED
    #define libusb_handle_events_timeout_completed(ctx, tx, completed) \
        libusb_handle_events_timeout(ctx, tx)
#endif

//! libusb_error_name is only in newer API
#ifndef HAVE_LIBUSB_ERROR_NAME
    #define libusb_error_name(code) \
        str(boost::format("LIBUSB_ERROR_CODE %d") % code)
#endif

//! type for sharing the release queue with managed buffers
class libusb_zero_copy_mb;
typedef boost::shared_ptr<bounded_buffer<libusb_zero_copy_mb *> > mb_queue_sptr;

/*!
 * The libusb docs state that status and actual length can only be read in the callback.
 * Therefore, this struct is intended to store data seen from the callback function.
 */
struct lut_result_t
{
    lut_result_t(void)
    {
        completed = 0;
        status = LIBUSB_TRANSFER_COMPLETED;
        actual_length = 0;
    }
    int completed;
    libusb_transfer_status status;
    int actual_length;
    boost::mutex mut;
    boost::condition_variable usb_transfer_complete;
};

// Created to be used as an argument to boost::condition_variable::timed_wait() function
struct lut_result_completed {
    const lut_result_t& _result;
    lut_result_completed(const lut_result_t& result):_result(result) {}
    bool operator()() const {return (_result.completed ? true : false);}
};

/*!
 * All libusb callback functions should be marked with the LIBUSB_CALL macro
 * to ensure that they are compiled with the same calling convention as libusb.
 */

//! helper function: handles all async callbacks
static void LIBUSB_CALL libusb_async_cb(libusb_transfer *lut)
{
    lut_result_t *r = (lut_result_t *)lut->user_data;
    boost::lock_guard<boost::mutex> lock(r->mut);
    r->status = lut->status;
    r->actual_length = lut->actual_length;
    r->completed = 1;
    r->usb_transfer_complete.notify_one();  // wake up thread waiting in wait_for_completion() member function below
}

/***********************************************************************
 * Reusable managed buffer:
 *  - Associated with a particular libusb transfer struct.
 *  - Submits the transfer to libusb in the release method.
 **********************************************************************/
class libusb_zero_copy_mb : public managed_buffer
{
public:
    libusb_zero_copy_mb(libusb_transfer *lut, const size_t frame_size, boost::function<void(libusb_zero_copy_mb *)> release_cb, const bool is_recv, const std::string &name):
        _release_cb(release_cb), _is_recv(is_recv), _name(name),
        _ctx(libusb::session::get_global_session()->get_context()),
        _lut(lut), _frame_size(frame_size) { /* NOP */ }

    void release(void){_release_cb(this);}

    UHD_INLINE void submit(void)
    {
        _lut->length = (_is_recv)? _frame_size : size(); //always set length
        const int ret = libusb_submit_transfer(_lut);
        if (ret != 0) throw uhd::runtime_error(str(boost::format(
            "usb %s submit failed: %s") % _name % libusb_error_name(ret)));
    }

    template <typename buffer_type>
    UHD_INLINE typename buffer_type::sptr get_new(const double timeout)
    {
        if (wait_for_completion(timeout))
        {
            if (result.status != LIBUSB_TRANSFER_COMPLETED) throw uhd::runtime_error(str(boost::format(
                "usb %s transfer status: %d") % _name % int(result.status)));
            result.completed = 0;
            return make(reinterpret_cast<buffer_type *>(this), _lut->buffer, (_is_recv)? result.actual_length : _frame_size);
        }
        return typename buffer_type::sptr();
    }

    // This is public because it is accessed from the libusb_zero_copy_single constructor
    lut_result_t result;

    /*!
     * Wait for a managed buffer to become complete.
     *
     * \param timeout the wait timeout in seconds.  A negative value will wait forever.
     * \return true for completion, false for timeout
     */
    UHD_INLINE bool wait_for_completion(const double timeout)
    {
        boost::unique_lock<boost::mutex> lock(result.mut);
        if (!result.completed) {
            if (timeout < 0.0) {
                result.usb_transfer_complete.wait(lock);
            } else {
                const boost::system_time timeout_time = boost::get_system_time() + boost::posix_time::microseconds(long(timeout*1000000));
                result.usb_transfer_complete.timed_wait(lock, timeout_time, lut_result_completed(result));
            }
        }
        return result.completed;
    }

private:

    boost::function<void(libusb_zero_copy_mb *)> _release_cb;
    const bool _is_recv;
    const std::string _name;
    libusb_context *_ctx;
    libusb_transfer *_lut;
    const size_t _frame_size;
};

/***********************************************************************
 * USB zero_copy device class
 **********************************************************************/
class libusb_zero_copy_single
{
public:
    libusb_zero_copy_single(
        libusb::device_handle::sptr handle,
        const size_t interface, const size_t endpoint,
        const size_t num_frames, const size_t frame_size
    ):
        _handle(handle),
        _num_frames(num_frames),
        _frame_size(frame_size),
        _buffer_pool(buffer_pool::make(_num_frames, _frame_size)),
        _enqueued(_num_frames), _released(_num_frames)
    {
        const bool is_recv = (endpoint & 0x80) != 0;
        const std::string name = str(boost::format("%s%d") % ((is_recv)? "rx" : "tx") % int(endpoint & 0x7f));
        _handle->claim_interface(interface);

        //flush the buffers out of the recv endpoint
        //limit the flushing to at most one second
        if (is_recv) for (size_t i = 0; i < 100; i++)
        {
            unsigned char buff[512];
            int transfered = 0;
            const int status = libusb_bulk_transfer(
                _handle->get(), // dev_handle
                endpoint, // endpoint
                static_cast<unsigned char *>(buff),
                sizeof(buff),
                &transfered, //bytes xfered
                10 //timeout ms
            );
            if (status == LIBUSB_ERROR_TIMEOUT) break;
        }

        //allocate libusb transfer structs and managed buffers
        for (size_t i = 0; i < get_num_frames(); i++)
        {
            libusb_transfer *lut = libusb_alloc_transfer(0);
            UHD_ASSERT_THROW(lut != NULL);

            _mb_pool.push_back(boost::make_shared<libusb_zero_copy_mb>(
                lut, this->get_frame_size(), boost::bind(&libusb_zero_copy_single::enqueue_damn_buffer, this, _1), is_recv, name
            ));

            libusb_fill_bulk_transfer(
                lut,                                                    // transfer
                _handle->get(),                                         // dev_handle
                endpoint,                                               // endpoint
                static_cast<unsigned char *>(_buffer_pool->at(i)),      // buffer
                this->get_frame_size(),                                 // length
                libusb_transfer_cb_fn(&libusb_async_cb),                // callback
                static_cast<void *>(&_mb_pool.back()->result),          // user_data
                0                                                       // timeout (ms)
            );

            _all_luts.push_back(lut);
        }

        //initial release for all buffers
        for (size_t i = 0; i < get_num_frames(); i++)
        {
            libusb_zero_copy_mb &mb = *(_mb_pool[i]);
            if (is_recv) mb.release();
            else
            {
                mb.result.completed = 1;
                _enqueued.push_back(&mb);
            }
        }
    }

    ~libusb_zero_copy_single(void)
    {
        //cancel all transfers
        BOOST_FOREACH(libusb_transfer *lut, _all_luts)
        {
            libusb_cancel_transfer(lut);
        }

        //process all transfers until timeout occurs
        BOOST_FOREACH(libusb_zero_copy_mb *mb, _enqueued)
        {
            mb->wait_for_completion(0.01);
        }

        //free all transfers
        BOOST_FOREACH(libusb_transfer *lut, _all_luts)
        {
            libusb_free_transfer(lut);
        }
    }

    template <typename buffer_type>
    UHD_INLINE typename buffer_type::sptr get_buff(double timeout)
    {
        typename buffer_type::sptr buff;
        libusb_zero_copy_mb *front = NULL;
        boost::mutex::scoped_lock lock(_mutex);
        if (_enqueued.empty())
        {
            _cond.timed_wait(lock, boost::posix_time::microseconds(long(timeout*1e6)));
        }
        if (_enqueued.empty()) return buff;
        front = _enqueued.front();

        lock.unlock();
        buff = front->get_new<buffer_type>(timeout);
        lock.lock();

        if (buff) _enqueued.pop_front();
        this->submit_what_we_can();
        return buff;
    }

    UHD_INLINE size_t get_num_frames(void) const { return _num_frames; }
    UHD_INLINE size_t get_frame_size(void) const { return _frame_size; }

private:
    libusb::device_handle::sptr _handle;
    const size_t _num_frames, _frame_size;

    //! Storage for transfer related objects
    buffer_pool::sptr _buffer_pool;
    std::vector<boost::shared_ptr<libusb_zero_copy_mb> > _mb_pool;

    boost::mutex _mutex;
    boost::condition_variable _cond;

    //! why 2 queues? there is room in the future to have > N buffers but only N in flight
    boost::circular_buffer<libusb_zero_copy_mb *> _enqueued, _released;

    void enqueue_damn_buffer(libusb_zero_copy_mb *mb)
    {
        boost::mutex::scoped_lock l(_mutex);
        _released.push_back(mb);
        this->submit_what_we_can();
        l.unlock();
        _cond.notify_one();
    }

    void submit_what_we_can(void)
    {
        while (not _released.empty() and not _enqueued.full())
        {
            _released.front()->submit();
            _enqueued.push_back(_released.front());
            _released.pop_front();
        }
    }

    //! a list of all transfer structs we allocated
    std::list<libusb_transfer *> _all_luts;
};

/***********************************************************************
 * USB zero_copy device class
 **********************************************************************/
struct libusb_zero_copy_impl : usb_zero_copy
{
    libusb_zero_copy_impl(
        libusb::device_handle::sptr handle,
        const size_t recv_interface,
        const size_t recv_endpoint,
        const size_t send_interface,
        const size_t send_endpoint,
        const device_addr_t &hints
    ){
        _recv_impl.reset(new libusb_zero_copy_single(
            handle, recv_interface, (recv_endpoint & 0x7f) | 0x80,
            size_t(hints.cast<double>("num_recv_frames", DEFAULT_NUM_XFERS)),
            size_t(hints.cast<double>("recv_frame_size", DEFAULT_XFER_SIZE))));
        _send_impl.reset(new libusb_zero_copy_single(
            handle, send_interface, (send_endpoint & 0x7f) | 0x00,
            size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_XFERS)),
            size_t(hints.cast<double>("send_frame_size", DEFAULT_XFER_SIZE))));
    }

    managed_recv_buffer::sptr get_recv_buff(double timeout)
    {
        boost::mutex::scoped_lock l(_recv_mutex);
        return _recv_impl->get_buff<managed_recv_buffer>(timeout);
    }

    managed_send_buffer::sptr get_send_buff(double timeout)
    {
        boost::mutex::scoped_lock l(_send_mutex);
        return _send_impl->get_buff<managed_send_buffer>(timeout);
    }

    size_t get_num_recv_frames(void) const { return _recv_impl->get_num_frames(); }
    size_t get_num_send_frames(void) const { return _send_impl->get_num_frames(); }

    size_t get_recv_frame_size(void) const { return _recv_impl->get_frame_size(); }
    size_t get_send_frame_size(void) const { return _send_impl->get_frame_size(); }

    boost::shared_ptr<libusb_zero_copy_single> _recv_impl, _send_impl;
    boost::mutex _recv_mutex, _send_mutex;
};

/***********************************************************************
 * USB zero_copy make functions
 **********************************************************************/
usb_zero_copy::sptr usb_zero_copy::make(
    usb_device_handle::sptr handle,
    const size_t recv_interface,
    const size_t recv_endpoint,
    const size_t send_interface,
    const 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<libusb::special_handle>(handle)->get_device()
    ));
    return sptr(new libusb_zero_copy_impl(
        dev_handle, recv_interface, recv_endpoint, send_interface, send_endpoint, hints
    ));
}