//
// Copyright 2010-2011 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
#include
#include
#include
#include
#include
#include
#include
#include
#include
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*/
/*!
* 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 rx async callbacks
static void LIBUSB_CALL libusb_async_rx_cb(libusb_transfer *lut){
if(lut->actual_length == 0) {
UHD_ASSERT_THROW(libusb_submit_transfer(lut) == 0); //get out until you find some real data
return;
}
(*static_cast *>(lut->user_data))();
}
//! helper function: handles all tx async callbacks
static void LIBUSB_CALL libusb_async_tx_cb(libusb_transfer *lut) {
(*static_cast *>(lut->user_data))();
}
//! callback to free transfer upon cancellation
static void LIBUSB_CALL cancel_transfer_cb(libusb_transfer *lut){
if (lut->status == LIBUSB_TRANSFER_CANCELLED || lut->status == LIBUSB_TRANSFER_TIMED_OUT) libusb_free_transfer(lut);
else UHD_MSG(error) << "libusb cancel_transfer unexpected status " << lut->status << std::endl;
}
/***********************************************************************
* Reusable managed receiver buffer:
* - Associated with a particular libusb transfer struct.
* - Submits the transfer to libusb in the release method.
**********************************************************************/
class libusb_zero_copy_mrb : public managed_recv_buffer{
public:
libusb_zero_copy_mrb(libusb_transfer *lut):
_lut(lut), _expired(true) { /* NOP */ }
void release(void){
if (_expired) return;
UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
_expired = true;
}
sptr get_new(void){
_expired = false;
return make_managed_buffer(this);
}
private:
const void *get_buff(void) const{return _lut->buffer;}
size_t get_size(void) const{return _lut->actual_length;}
libusb_transfer *_lut;
bool _expired;
};
/***********************************************************************
* Reusable managed send buffer:
* - Associated with a particular libusb transfer struct.
* - Submits the transfer to libusb in the commit method.
**********************************************************************/
class libusb_zero_copy_msb : public managed_send_buffer{
public:
libusb_zero_copy_msb(libusb_transfer *lut):
_lut(lut), _expired(true) { /* NOP */ }
void commit(size_t len){
if (_expired) return;
_lut->length = len;
if(len == 0) libusb_async_tx_cb(_lut);
else UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
_expired = true;
}
sptr get_new(void){
_expired = false;
return make_managed_buffer(this);
}
private:
void *get_buff(void) const{return _lut->buffer;}
size_t get_size(void) const{return _lut->length;}
libusb_transfer *_lut;
bool _expired;
};
/***********************************************************************
* USB zero_copy device class
**********************************************************************/
class libusb_zero_copy_impl : public usb_zero_copy{
public:
libusb_zero_copy_impl(
libusb::device_handle::sptr handle,
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))),
_recv_buffer_pool(buffer_pool::make(_num_recv_frames, _recv_frame_size)),
_send_buffer_pool(buffer_pool::make(_num_send_frames, _send_frame_size)),
_pending_recv_buffs(_num_recv_frames),
_pending_send_buffs(_num_send_frames)
{
_handle->claim_interface(2 /*in interface*/);
_handle->claim_interface(1 /*out interface*/);
//allocate libusb transfer structs and managed receive buffers
for (size_t i = 0; i < get_num_recv_frames(); i++){
libusb_transfer *lut = libusb_alloc_transfer(0);
UHD_ASSERT_THROW(lut != NULL);
_mrb_pool.push_back(libusb_zero_copy_mrb(lut));
_callbacks.push_back(boost::bind(
&libusb_zero_copy_impl::handle_recv, this, &_mrb_pool.back()
));
libusb_fill_bulk_transfer(
lut, // transfer
_handle->get(), // dev_handle
(recv_endpoint & 0x7f) | 0x80, // endpoint
static_cast(_recv_buffer_pool->at(i)), // buffer
this->get_recv_frame_size(), // length
libusb_transfer_cb_fn(&libusb_async_rx_cb), // callback
static_cast(&_callbacks.back()), // user_data
0 // timeout (ms)
);
_all_luts.push_back(lut);
_mrb_pool.back().get_new();
}
//allocate libusb transfer structs and managed send buffers
for (size_t i = 0; i < get_num_send_frames(); i++){
libusb_transfer *lut = libusb_alloc_transfer(0);
UHD_ASSERT_THROW(lut != NULL);
_msb_pool.push_back(libusb_zero_copy_msb(lut));
_callbacks.push_back(boost::bind(
&libusb_zero_copy_impl::handle_send, this, &_msb_pool.back()
));
libusb_fill_bulk_transfer(
lut, // transfer
_handle->get(), // dev_handle
(send_endpoint & 0x7f) | 0x00, // endpoint
static_cast(_send_buffer_pool->at(i)), // buffer
this->get_send_frame_size(), // length
libusb_transfer_cb_fn(&libusb_async_tx_cb), // callback
static_cast(&_callbacks.back()), // user_data
0 // timeout
);
_all_luts.push_back(lut);
libusb_async_tx_cb(lut);
}
//spawn the event handler threads
const size_t concurrency = hints.cast("concurrency_hint", 1);
for (size_t i = 0; i < concurrency; i++) _event_loop_tasks.push_back(task::make(
boost::bind(&libusb_zero_copy_impl::run_event_loop, this)
));
}
~libusb_zero_copy_impl(void){
//cancel and free all transfers
BOOST_FOREACH(libusb_transfer *lut, _all_luts){
lut->callback = libusb_transfer_cb_fn(&cancel_transfer_cb);
libusb_cancel_transfer(lut);
while(lut->status != LIBUSB_TRANSFER_CANCELLED
&& lut->status != LIBUSB_TRANSFER_COMPLETED
&& lut->status != LIBUSB_TRANSFER_TIMED_OUT) {
boost::this_thread::sleep(boost::posix_time::milliseconds(10));
}
}
}
managed_recv_buffer::sptr get_recv_buff(double timeout){
libusb_zero_copy_mrb *mrb = NULL;
if (_pending_recv_buffs.pop_with_timed_wait(mrb, timeout)){
return mrb->get_new();
}
return managed_recv_buffer::sptr();
}
managed_send_buffer::sptr get_send_buff(double timeout){
libusb_zero_copy_msb *msb = NULL;
if (_pending_send_buffs.pop_with_timed_wait(msb, timeout)){
return msb->get_new();
}
return managed_send_buffer::sptr();
}
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:
//! Handle a bound async callback for recv
void handle_recv(libusb_zero_copy_mrb *mrb){
_pending_recv_buffs.push_with_haste(mrb);
}
//! Handle a bound async callback for send
void handle_send(libusb_zero_copy_msb *msb){
_pending_send_buffs.push_with_haste(msb);
}
libusb::device_handle::sptr _handle;
const size_t _recv_frame_size, _num_recv_frames;
const size_t _send_frame_size, _num_send_frames;
//! Storage for transfer related objects
buffer_pool::sptr _recv_buffer_pool, _send_buffer_pool;
bounded_buffer _pending_recv_buffs;
bounded_buffer _pending_send_buffs;
std::list _mrb_pool;
std::list _msb_pool;
std::list > _callbacks;
//! a list of all transfer structs we allocated
std::list _all_luts;
//! event handler threads
std::list _event_loop_tasks;
void run_event_loop(void){
set_thread_priority_safe();
libusb_context *context = libusb::session::get_global_session()->get_context();
while (not boost::this_thread::interruption_requested()){
timeval tv;
tv.tv_sec = 0;
tv.tv_usec = 100000; //100ms
libusb_handle_events_timeout(context, &tv);
}
}
};
/***********************************************************************
* USB zero_copy make functions
**********************************************************************/
usb_zero_copy::sptr usb_zero_copy::make(
usb_device_handle::sptr handle,
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, hints
));
}