aboutsummaryrefslogtreecommitdiffstats
path: root/host/lib/transport
diff options
context:
space:
mode:
Diffstat (limited to 'host/lib/transport')
-rw-r--r--host/lib/transport/CMakeLists.txt6
-rw-r--r--host/lib/transport/libusb1_zero_copy.cpp47
-rw-r--r--host/lib/transport/simple_claimer.hpp64
-rw-r--r--host/lib/transport/super_send_packet_handler.hpp1
-rw-r--r--host/lib/transport/udp_wsa_zero_copy.cpp300
-rw-r--r--host/lib/transport/udp_zero_copy.cpp139
-rw-r--r--host/lib/transport/usb_zero_copy_wrapper.cpp104
7 files changed, 493 insertions, 168 deletions
diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt
index 8e8ea5ea8..a95560770 100644
--- a/host/lib/transport/CMakeLists.txt
+++ b/host/lib/transport/CMakeLists.txt
@@ -82,7 +82,11 @@ SET_SOURCE_FILES_PROPERTIES(
########################################################################
# Setup UDP
########################################################################
-LIBUHD_APPEND_SOURCES(${CMAKE_CURRENT_SOURCE_DIR}/udp_zero_copy.cpp)
+IF(WIN32)
+ LIBUHD_APPEND_SOURCES(${CMAKE_CURRENT_SOURCE_DIR}/udp_wsa_zero_copy.cpp)
+ELSE()
+ LIBUHD_APPEND_SOURCES(${CMAKE_CURRENT_SOURCE_DIR}/udp_zero_copy.cpp)
+ENDIF()
#On windows, the boost asio implementation uses the winsock2 library.
#Note: we exclude the .lib extension for cygwin and mingw platforms.
diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index 3e67264cd..c13384eec 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -21,6 +21,7 @@
#include <uhd/utils/msg.hpp>
#include <uhd/exception.hpp>
#include <boost/foreach.hpp>
+#include <boost/make_shared.hpp>
#include <boost/thread/thread.hpp>
#include <list>
@@ -61,8 +62,18 @@ static void LIBUSB_CALL libusb_async_cb(libusb_transfer *lut){
* \return true for completion, false for timeout
*/
UHD_INLINE bool wait_for_completion(libusb_context *ctx, const double timeout, bool &completed){
- const boost::system_time timeout_time = boost::get_system_time() + boost::posix_time::microseconds(long(timeout*1000000));
+ //already completed by a previous call?
+ if (completed) return true;
+
+ //perform a non-blocking event handle
+ timeval tv;
+ tv.tv_sec = 0;
+ tv.tv_usec = 0;
+ libusb_handle_events_timeout(ctx, &tv);
+ if (completed) return true;
+ //finish the rest with a timeout loop
+ const boost::system_time timeout_time = boost::get_system_time() + boost::posix_time::microseconds(long(timeout*1000000));
while (not completed and (boost::get_system_time() < timeout_time)){
timeval tv;
tv.tv_sec = 0;
@@ -82,21 +93,18 @@ class libusb_zero_copy_mrb : public managed_recv_buffer{
public:
libusb_zero_copy_mrb(libusb_transfer *lut, const size_t frame_size):
_ctx(libusb::session::get_global_session()->get_context()),
- _lut(lut), _expired(false), _frame_size(frame_size) { /* NOP */ }
+ _lut(lut), _frame_size(frame_size) { /* NOP */ }
void release(void){
- if (_expired) return;
completed = false;
_lut->length = _frame_size; //always reset length
UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
- _expired = true;
}
sptr get_new(const double timeout, size_t &index){
if (wait_for_completion(_ctx, timeout, completed)){
index++;
- _expired = false;
- return make_managed_buffer(this);
+ return make(this, _lut->buffer, _lut->actual_length);
}
return managed_recv_buffer::sptr();
}
@@ -104,12 +112,8 @@ public:
bool completed;
private:
- const void *get_buff(void) const{return _lut->buffer;}
- size_t get_size(void) const{return _lut->actual_length;}
-
libusb_context *_ctx;
libusb_transfer *_lut;
- bool _expired;
const size_t _frame_size;
};
@@ -122,22 +126,18 @@ class libusb_zero_copy_msb : public managed_send_buffer{
public:
libusb_zero_copy_msb(libusb_transfer *lut, const size_t frame_size):
_ctx(libusb::session::get_global_session()->get_context()),
- _lut(lut), _expired(false), _frame_size(frame_size) { /* NOP */ }
+ _lut(lut), _frame_size(frame_size) { completed = true; }
- void commit(size_t len){
- if (_expired) return;
+ void release(void){
completed = false;
- _lut->length = len;
- if (len == 0) libusb_async_cb(_lut);
- else UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
- _expired = true;
+ _lut->length = size();
+ UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
}
sptr get_new(const double timeout, size_t &index){
if (wait_for_completion(_ctx, timeout, completed)){
index++;
- _expired = false;
- return make_managed_buffer(this);
+ return make(this, _lut->buffer, _frame_size);
}
return managed_send_buffer::sptr();
}
@@ -145,12 +145,8 @@ public:
bool completed;
private:
- void *get_buff(void) const{return _lut->buffer;}
- size_t get_size(void) const{return _frame_size;}
-
libusb_context *_ctx;
libusb_transfer *_lut;
- bool _expired;
const size_t _frame_size;
};
@@ -187,7 +183,7 @@ public:
libusb_transfer *lut = libusb_alloc_transfer(0);
UHD_ASSERT_THROW(lut != NULL);
- _mrb_pool.push_back(boost::shared_ptr<libusb_zero_copy_mrb>(new libusb_zero_copy_mrb(lut, this->get_recv_frame_size())));
+ _mrb_pool.push_back(boost::make_shared<libusb_zero_copy_mrb>(lut, this->get_recv_frame_size()));
libusb_fill_bulk_transfer(
lut, // transfer
@@ -210,7 +206,7 @@ public:
libusb_transfer *lut = libusb_alloc_transfer(0);
UHD_ASSERT_THROW(lut != NULL);
- _msb_pool.push_back(boost::shared_ptr<libusb_zero_copy_msb>(new libusb_zero_copy_msb(lut, this->get_send_frame_size())));
+ _msb_pool.push_back(boost::make_shared<libusb_zero_copy_msb>(lut, this->get_send_frame_size()));
libusb_fill_bulk_transfer(
lut, // transfer
@@ -224,7 +220,6 @@ public:
);
_all_luts.push_back(lut);
- _msb_pool.back()->commit(0);
}
}
diff --git a/host/lib/transport/simple_claimer.hpp b/host/lib/transport/simple_claimer.hpp
new file mode 100644
index 000000000..3bbc49a05
--- /dev/null
+++ b/host/lib/transport/simple_claimer.hpp
@@ -0,0 +1,64 @@
+//
+// Copyright 2012 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/>.
+//
+
+#ifndef INCLUDED_LIBUHD_TRANSPORT_SIMPLE_CLAIMER_HPP
+#define INCLUDED_LIBUHD_TRANSPORT_SIMPLE_CLAIMER_HPP
+
+#include <uhd/config.hpp>
+#include <boost/thread/condition.hpp>
+#include <boost/thread/mutex.hpp>
+
+namespace uhd{ namespace transport{
+
+/***********************************************************************
+ * Claimer class to provide synchronization for multi-thread access.
+ * Claiming enables buffer classes to be used with a buffer queue.
+ **********************************************************************/
+class simple_claimer{
+public:
+ simple_claimer(void){
+ this->release();
+ }
+
+ UHD_INLINE void release(void){
+ boost::mutex::scoped_lock lock(_mutex);
+ _locked = false;
+ lock.unlock();
+ _cond.notify_one();
+ }
+
+ UHD_INLINE bool claim_with_wait(const double timeout){
+ boost::mutex::scoped_lock lock(_mutex);
+ while (_locked){
+ if (not _cond.timed_wait(lock, boost::posix_time::microseconds(long(timeout*1e6)))){
+ break;
+ }
+ }
+ const bool ret = not _locked;
+ _locked = true;
+ return ret;
+ }
+
+private:
+ bool _locked;
+ boost::mutex _mutex;
+ boost::condition_variable _cond;
+};
+
+}} //namespace uhd::transport
+
+#endif /* INCLUDED_LIBUHD_TRANSPORT_SIMPLE_CLAIMER_HPP */
diff --git a/host/lib/transport/super_send_packet_handler.hpp b/host/lib/transport/super_send_packet_handler.hpp
index 46c98afea..8a4fce11e 100644
--- a/host/lib/transport/super_send_packet_handler.hpp
+++ b/host/lib/transport/super_send_packet_handler.hpp
@@ -242,6 +242,7 @@ private:
//commit the samples to the zero-copy interface
size_t num_bytes_total = (_header_offset_words32+if_packet_info.num_packet_words32)*sizeof(boost::uint32_t);
buff->commit(num_bytes_total);
+ buff.reset(); //effectively a release
}
_next_packet_seq++; //increment sequence after commits
diff --git a/host/lib/transport/udp_wsa_zero_copy.cpp b/host/lib/transport/udp_wsa_zero_copy.cpp
new file mode 100644
index 000000000..6fe4e3cad
--- /dev/null
+++ b/host/lib/transport/udp_wsa_zero_copy.cpp
@@ -0,0 +1,300 @@
+//
+// 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 <http://www.gnu.org/licenses/>.
+//
+
+#include "udp_common.hpp"
+#include <uhd/transport/udp_zero_copy.hpp>
+#include <uhd/transport/udp_simple.hpp> //mtu
+#include <uhd/transport/buffer_pool.hpp>
+#include <uhd/utils/msg.hpp>
+#include <uhd/utils/log.hpp>
+#include <boost/format.hpp>
+#include <vector>
+
+using namespace uhd;
+using namespace uhd::transport;
+namespace asio = boost::asio;
+
+//A reasonable number of frames for send/recv and async/sync
+static const size_t DEFAULT_NUM_FRAMES = 32;
+
+/***********************************************************************
+ * Check registry for correct fast-path setting (windows only)
+ **********************************************************************/
+#ifdef HAVE_ATLBASE_H
+#define CHECK_REG_SEND_THRESH
+#include <atlbase.h> //CRegKey
+static void check_registry_for_fast_send_threshold(const size_t mtu){
+ static bool warned = false;
+ if (warned) return; //only allow one printed warning per process
+
+ CRegKey reg_key;
+ DWORD threshold = 1024; //system default when threshold is not specified
+ if (
+ reg_key.Open(HKEY_LOCAL_MACHINE, "System\\CurrentControlSet\\Services\\AFD\\Parameters", KEY_READ) != ERROR_SUCCESS or
+ reg_key.QueryDWORDValue("FastSendDatagramThreshold", threshold) != ERROR_SUCCESS or threshold < mtu
+ ){
+ UHD_MSG(warning) << boost::format(
+ "The MTU (%d) is larger than the FastSendDatagramThreshold (%d)!\n"
+ "This will negatively affect the transmit performance.\n"
+ "See the transport application notes for more detail.\n"
+ ) % mtu % threshold << std::endl;
+ warned = true;
+ }
+ reg_key.Close();
+}
+#endif /*HAVE_ATLBASE_H*/
+
+/***********************************************************************
+ * Static initialization to take care of WSA init and cleanup
+ **********************************************************************/
+struct uhd_wsa_control{
+ uhd_wsa_control(void){
+ WSADATA wsaData;
+ WSAStartup(MAKEWORD(2, 2), &wsaData); /*windows socket startup */
+ }
+
+ ~uhd_wsa_control(void){
+ WSACleanup();
+ }
+};
+
+/***********************************************************************
+ * Reusable managed receiver buffer:
+ * - Initialize with memory and a release callback.
+ * - Call get new with a length in bytes to re-use.
+ **********************************************************************/
+class udp_zero_copy_asio_mrb : public managed_recv_buffer{
+public:
+ udp_zero_copy_asio_mrb(void *mem, int sock_fd, const size_t frame_size):
+ _sock_fd(sock_fd), _frame_size(frame_size)
+ {
+ _wsa_buff.buf = reinterpret_cast<char *>(mem);
+ ZeroMemory(&_overlapped, sizeof(_overlapped));
+ _overlapped.hEvent = WSACreateEvent();
+ UHD_ASSERT_THROW(_overlapped.hEvent != WSA_INVALID_EVENT);
+ this->release(); //makes buffer available via get_new
+ }
+
+ ~udp_zero_copy_asio_mrb(void){
+ WSACloseEvent(_overlapped.hEvent);
+ }
+
+ void release(void){
+ _wsa_buff.len = _frame_size;
+ _flags = 0;
+ WSARecv(_sock_fd, &_wsa_buff, 1, &_wsa_buff.len, &_flags, &_overlapped, NULL);
+ }
+
+ UHD_INLINE sptr get_new(const double timeout, size_t &index){
+ const DWORD result = WSAWaitForMultipleEvents(
+ 1, &_overlapped.hEvent, true, DWORD(timeout*1000), true
+ );
+ if (result == WSA_WAIT_TIMEOUT) return managed_recv_buffer::sptr();
+ index++; //advances the caller's buffer
+
+ WSAGetOverlappedResult(_sock_fd, &_overlapped, &_wsa_buff.len, true, &_flags);
+
+ WSAResetEvent(_overlapped.hEvent);
+ return make(this, _wsa_buff.buf, _wsa_buff.len);
+ }
+
+private:
+ int _sock_fd;
+ const size_t _frame_size;
+ WSAOVERLAPPED _overlapped;
+ WSABUF _wsa_buff;
+ DWORD _flags;
+};
+
+/***********************************************************************
+ * Reusable managed send buffer:
+ * - committing the buffer calls the asynchronous socket send
+ * - getting a new buffer performs the blocking wait for completion
+ **********************************************************************/
+class udp_zero_copy_asio_msb : public managed_send_buffer{
+public:
+ udp_zero_copy_asio_msb(void *mem, int sock_fd, const size_t frame_size):
+ _sock_fd(sock_fd), _frame_size(frame_size)
+ {
+ _wsa_buff.buf = reinterpret_cast<char *>(mem);
+ ZeroMemory(&_overlapped, sizeof(_overlapped));
+ _overlapped.hEvent = WSACreateEvent();
+ UHD_ASSERT_THROW(_overlapped.hEvent != WSA_INVALID_EVENT);
+ WSASetEvent(_overlapped.hEvent); //makes buffer available via get_new
+ }
+
+ ~udp_zero_copy_asio_msb(void){
+ WSACloseEvent(_overlapped.hEvent);
+ }
+
+ void release(void){
+ _wsa_buff.len = size();
+ WSASend(_sock_fd, &_wsa_buff, 1, NULL, 0, &_overlapped, NULL);
+ }
+
+ UHD_INLINE sptr get_new(const double timeout, size_t &index){
+ const DWORD result = WSAWaitForMultipleEvents(
+ 1, &_overlapped.hEvent, true, DWORD(timeout*1000), true
+ );
+ if (result == WSA_WAIT_TIMEOUT) return managed_send_buffer::sptr();
+ index++; //advances the caller's buffer
+
+ WSAResetEvent(_overlapped.hEvent);
+ _wsa_buff.len = _frame_size;
+ return make(this, _wsa_buff.buf, _wsa_buff.len);
+ }
+
+private:
+ int _sock_fd;
+ const size_t _frame_size;
+ WSAOVERLAPPED _overlapped;
+ WSABUF _wsa_buff;
+};
+
+/***********************************************************************
+ * Zero Copy UDP implementation with WSA:
+ *
+ * This is not a true zero copy implementation as each
+ * send and recv requires a copy operation to/from userspace.
+ *
+ * For receive, use a blocking recv() call on the socket.
+ * This has better performance than the overlapped IO.
+ * For send, use overlapped IO to submit async sends.
+ **********************************************************************/
+class udp_zero_copy_wsa_impl : public udp_zero_copy{
+public:
+ typedef boost::shared_ptr<udp_zero_copy_wsa_impl> sptr;
+
+ udp_zero_copy_wsa_impl(
+ const std::string &addr,
+ const std::string &port,
+ const device_addr_t &hints
+ ):
+ _recv_frame_size(size_t(hints.cast<double>("recv_frame_size", udp_simple::mtu))),
+ _num_recv_frames(size_t(hints.cast<double>("num_recv_frames", DEFAULT_NUM_FRAMES))),
+ _send_frame_size(size_t(hints.cast<double>("send_frame_size", udp_simple::mtu))),
+ _num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_FRAMES))),
+ _recv_buffer_pool(buffer_pool::make(_num_recv_frames, _recv_frame_size)),
+ _send_buffer_pool(buffer_pool::make(_num_send_frames, _send_frame_size)),
+ _next_recv_buff_index(0), _next_send_buff_index(0)
+ {
+ #ifdef CHECK_REG_SEND_THRESH
+ check_registry_for_fast_send_threshold(this->get_send_frame_size());
+ #endif /*CHECK_REG_SEND_THRESH*/
+
+ UHD_MSG(status) << boost::format("Creating WSA UDP transport for %s:%s") % addr % port << std::endl;
+ static uhd_wsa_control uhd_wsa; //makes wsa start happen via lazy initialization
+
+ UHD_ASSERT_THROW(_num_send_frames <= WSA_MAXIMUM_WAIT_EVENTS);
+
+ //resolve the address
+ asio::io_service io_service;
+ 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 the socket
+ _sock_fd = WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_OVERLAPPED);
+ if (_sock_fd == INVALID_SOCKET){
+ const DWORD error = WSAGetLastError();
+ throw uhd::os_error(str(boost::format("WSASocket() failed with error %d") % error));
+ }
+
+ //set the socket non-blocking for recv
+ //u_long mode = 1;
+ //ioctlsocket(_sock_fd, FIONBIO, &mode);
+
+ //resize the socket buffers
+ const int recv_buff_size = int(hints.cast<double>("recv_buff_size", 0.0));
+ const int send_buff_size = int(hints.cast<double>("send_buff_size", 0.0));
+ if (recv_buff_size > 0) setsockopt(_sock_fd, SOL_SOCKET, SO_RCVBUF, (const char *)&recv_buff_size, sizeof(recv_buff_size));
+ if (send_buff_size > 0) setsockopt(_sock_fd, SOL_SOCKET, SO_SNDBUF, (const char *)&send_buff_size, sizeof(send_buff_size));
+
+ //connect the socket so we can send/recv
+ const asio::ip::udp::endpoint::data_type &servaddr = *receiver_endpoint.data();
+ if (WSAConnect(_sock_fd, (const struct sockaddr *)&servaddr, sizeof(servaddr), NULL, NULL, NULL, NULL) != 0){
+ const DWORD error = WSAGetLastError();
+ closesocket(_sock_fd);
+ throw uhd::os_error(str(boost::format("WSAConnect() failed with error %d") % error));
+ }
+
+ //allocate re-usable managed receive buffers
+ for (size_t i = 0; i < get_num_recv_frames(); i++){
+ _mrb_pool.push_back(boost::shared_ptr<udp_zero_copy_asio_mrb>(
+ new udp_zero_copy_asio_mrb(_recv_buffer_pool->at(i), _sock_fd, get_recv_frame_size())
+ ));
+ }
+
+ //allocate re-usable managed send buffers
+ for (size_t i = 0; i < get_num_send_frames(); i++){
+ _msb_pool.push_back(boost::shared_ptr<udp_zero_copy_asio_msb>(
+ new udp_zero_copy_asio_msb(_send_buffer_pool->at(i), _sock_fd, get_send_frame_size())
+ ));
+ }
+ }
+
+ ~udp_zero_copy_wsa_impl(void){
+ closesocket(_sock_fd);
+ }
+
+ /*******************************************************************
+ * Receive implementation:
+ * Block on the managed buffer's get call and advance the index.
+ ******************************************************************/
+ managed_recv_buffer::sptr get_recv_buff(double timeout){
+ if (_next_recv_buff_index == _num_recv_frames) _next_recv_buff_index = 0;
+ return _mrb_pool[_next_recv_buff_index]->get_new(timeout, _next_recv_buff_index);
+ }
+
+ size_t get_num_recv_frames(void) const {return _num_recv_frames;}
+ size_t get_recv_frame_size(void) const {return _recv_frame_size;}
+
+ /*******************************************************************
+ * Send implementation:
+ * Block on the managed buffer's get call and advance the index.
+ ******************************************************************/
+ managed_send_buffer::sptr get_send_buff(double timeout){
+ if (_next_send_buff_index == _num_send_frames) _next_send_buff_index = 0;
+ return _msb_pool[_next_send_buff_index]->get_new(timeout, _next_send_buff_index);
+ }
+
+ 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:
+ //memory management -> buffers and fifos
+ const size_t _recv_frame_size, _num_recv_frames;
+ const size_t _send_frame_size, _num_send_frames;
+ buffer_pool::sptr _recv_buffer_pool, _send_buffer_pool;
+ std::vector<boost::shared_ptr<udp_zero_copy_asio_msb> > _msb_pool;
+ std::vector<boost::shared_ptr<udp_zero_copy_asio_mrb> > _mrb_pool;
+ size_t _next_recv_buff_index, _next_send_buff_index;
+
+ //socket guts
+ SOCKET _sock_fd;
+};
+
+/***********************************************************************
+ * UDP zero copy make function
+ **********************************************************************/
+udp_zero_copy::sptr udp_zero_copy::make(
+ const std::string &addr,
+ const std::string &port,
+ const device_addr_t &hints
+){
+ return sptr(new udp_zero_copy_wsa_impl(addr, port, hints));
+}
diff --git a/host/lib/transport/udp_zero_copy.cpp b/host/lib/transport/udp_zero_copy.cpp
index 0ccc92b82..9765c19c0 100644
--- a/host/lib/transport/udp_zero_copy.cpp
+++ b/host/lib/transport/udp_zero_copy.cpp
@@ -1,5 +1,5 @@
//
-// Copyright 2010-2011 Ettus Research LLC
+// Copyright 2010-2012 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
@@ -16,14 +16,15 @@
//
#include "udp_common.hpp"
+#include "simple_claimer.hpp"
#include <uhd/transport/udp_zero_copy.hpp>
#include <uhd/transport/udp_simple.hpp> //mtu
-#include <uhd/transport/bounded_buffer.hpp>
#include <uhd/transport/buffer_pool.hpp>
#include <uhd/utils/msg.hpp>
#include <uhd/utils/log.hpp>
#include <boost/format.hpp>
-#include <list>
+#include <boost/make_shared.hpp>
+#include <vector>
using namespace uhd;
using namespace uhd::transport;
@@ -61,66 +62,71 @@ static void check_registry_for_fast_send_threshold(const size_t mtu){
/***********************************************************************
* Reusable managed receiver buffer:
- * - Initialize with memory and a release callback.
- * - Call get new with a length in bytes to re-use.
+ * - get_new performs the recv operation
**********************************************************************/
class udp_zero_copy_asio_mrb : public managed_recv_buffer{
public:
- udp_zero_copy_asio_mrb(void *mem, bounded_buffer<udp_zero_copy_asio_mrb *> &pending):
- _mem(mem), _len(0), _pending(pending){/* NOP */}
+ udp_zero_copy_asio_mrb(void *mem, int sock_fd, const size_t frame_size):
+ _mem(mem), _sock_fd(sock_fd), _frame_size(frame_size) { /*NOP*/ }
void release(void){
- if (_len == 0) return;
- _pending.push_with_haste(this);
- _len = 0;
+ _claimer.release();
}
- sptr get_new(size_t len){
- _len = len;
- return make_managed_buffer(this);
- }
+ UHD_INLINE sptr get_new(const double timeout, size_t &index){
+ if (not _claimer.claim_with_wait(timeout)) return sptr();
- template <class T> T cast(void) const{return static_cast<T>(_mem);}
+ #ifdef MSG_DONTWAIT //try a non-blocking recv() if supported
+ _len = ::recv(_sock_fd, (char *)_mem, _frame_size, MSG_DONTWAIT);
+ if (_len > 0){
+ index++; //advances the caller's buffer
+ return make(this, _mem, size_t(_len));
+ }
+ #endif
-private:
- const void *get_buff(void) const{return _mem;}
- size_t get_size(void) const{return _len;}
+ if (wait_for_recv_ready(_sock_fd, timeout)){
+ _len = ::recv(_sock_fd, (char *)_mem, _frame_size, 0);
+ index++; //advances the caller's buffer
+ return make(this, _mem, size_t(_len));
+ }
+ _claimer.release(); //undo claim
+ return sptr(); //null for timeout
+ }
+
+private:
void *_mem;
- size_t _len;
- bounded_buffer<udp_zero_copy_asio_mrb *> &_pending;
+ int _sock_fd;
+ size_t _frame_size;
+ ssize_t _len;
+ simple_claimer _claimer;
};
/***********************************************************************
* Reusable managed send buffer:
- * - Initialize with memory and a commit callback.
- * - Call get new with a length in bytes to re-use.
+ * - commit performs the send operation
**********************************************************************/
class udp_zero_copy_asio_msb : public managed_send_buffer{
public:
- udp_zero_copy_asio_msb(void *mem, bounded_buffer<udp_zero_copy_asio_msb *> &pending, int sock_fd):
- _mem(mem), _len(0), _pending(pending), _sock_fd(sock_fd){/* NOP */}
-
- void commit(size_t len){
- if (_len == 0) return;
- ::send(_sock_fd, this->cast<const char *>(), len, 0);
- _pending.push_with_haste(this);
- _len = 0;
+ udp_zero_copy_asio_msb(void *mem, int sock_fd, const size_t frame_size):
+ _mem(mem), _sock_fd(sock_fd), _frame_size(frame_size) { /*NOP*/ }
+
+ void release(void){
+ UHD_ASSERT_THROW(::send(_sock_fd, (const char *)_mem, size(), 0) == ssize_t(size()));
+ _claimer.release();
}
- sptr get_new(size_t len){
- _len = len;
- return make_managed_buffer(this);
+ UHD_INLINE sptr get_new(const double timeout, size_t &index){
+ if (not _claimer.claim_with_wait(timeout)) return sptr();
+ index++; //advances the caller's buffer
+ return make(this, _mem, _frame_size);
}
private:
- void *get_buff(void) const{return _mem;}
- size_t get_size(void) const{return _len;}
-
void *_mem;
- size_t _len;
- bounded_buffer<udp_zero_copy_asio_msb *> &_pending;
int _sock_fd;
+ size_t _frame_size;
+ simple_claimer _claimer;
};
/***********************************************************************
@@ -145,8 +151,7 @@ public:
_num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_FRAMES))),
_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)
+ _next_recv_buff_index(0), _next_send_buff_index(0)
{
UHD_LOG << boost::format("Creating udp transport for %s %s") % addr % port << std::endl;
@@ -167,18 +172,16 @@ public:
//allocate re-usable managed receive buffers
for (size_t i = 0; i < get_num_recv_frames(); i++){
- _mrb_pool.push_back(udp_zero_copy_asio_mrb(
- _recv_buffer_pool->at(i), _pending_recv_buffs
+ _mrb_pool.push_back(boost::make_shared<udp_zero_copy_asio_mrb>(
+ _recv_buffer_pool->at(i), _sock_fd, get_recv_frame_size()
));
- _pending_recv_buffs.push_with_haste(&_mrb_pool.back());
}
//allocate re-usable managed send buffers
for (size_t i = 0; i < get_num_send_frames(); i++){
- _msb_pool.push_back(udp_zero_copy_asio_msb(
- _send_buffer_pool->at(i), _pending_send_buffs, _sock_fd
+ _msb_pool.push_back(boost::make_shared<udp_zero_copy_asio_msb>(
+ _send_buffer_pool->at(i), _sock_fd, get_send_frame_size()
));
- _pending_send_buffs.push_with_haste(&_msb_pool.back());
}
}
@@ -198,29 +201,11 @@ public:
/*******************************************************************
* Receive implementation:
- *
- * Perform a non-blocking receive for performance,
- * and then fall back to a blocking receive with timeout.
- * Return the managed receive buffer with the new length.
- * When the caller is finished with the managed buffer,
- * the managed receive buffer is released back into the queue.
+ * Block on the managed buffer's get call and advance the index.
******************************************************************/
managed_recv_buffer::sptr get_recv_buff(double timeout){
- udp_zero_copy_asio_mrb *mrb = NULL;
- if (_pending_recv_buffs.pop_with_timed_wait(mrb, timeout)){
-
- #ifdef MSG_DONTWAIT //try a non-blocking recv() if supported
- ssize_t ret = ::recv(_sock_fd, mrb->cast<char *>(), _recv_frame_size, MSG_DONTWAIT);
- if (ret > 0) return mrb->get_new(ret);
- #endif
-
- if (wait_for_recv_ready(_sock_fd, timeout)) return mrb->get_new(
- ::recv(_sock_fd, mrb->cast<char *>(), _recv_frame_size, 0)
- );
-
- _pending_recv_buffs.push_with_haste(mrb); //timeout: return the managed buffer to the queue
- }
- return managed_recv_buffer::sptr();
+ if (_next_recv_buff_index == _num_recv_frames) _next_recv_buff_index = 0;
+ return _mrb_pool[_next_recv_buff_index]->get_new(timeout, _next_recv_buff_index);
}
size_t get_num_recv_frames(void) const {return _num_recv_frames;}
@@ -228,18 +213,11 @@ public:
/*******************************************************************
* Send implementation:
- *
- * Get a managed receive buffer immediately with max length set.
- * The caller will fill the buffer and commit it when finished.
- * The commit routine will perform a blocking send operation,
- * and push the managed send buffer back into the queue.
+ * Block on the managed buffer's get call and advance the index.
******************************************************************/
managed_send_buffer::sptr get_send_buff(double timeout){
- udp_zero_copy_asio_msb *msb = NULL;
- if (_pending_send_buffs.pop_with_timed_wait(msb, timeout)){
- return msb->get_new(_send_frame_size);
- }
- return managed_send_buffer::sptr();
+ if (_next_send_buff_index == _num_send_frames) _next_send_buff_index = 0;
+ return _msb_pool[_next_send_buff_index]->get_new(timeout, _next_send_buff_index);
}
size_t get_num_send_frames(void) const {return _num_send_frames;}
@@ -250,10 +228,9 @@ private:
const size_t _recv_frame_size, _num_recv_frames;
const size_t _send_frame_size, _num_send_frames;
buffer_pool::sptr _recv_buffer_pool, _send_buffer_pool;
- bounded_buffer<udp_zero_copy_asio_mrb *> _pending_recv_buffs;
- bounded_buffer<udp_zero_copy_asio_msb *> _pending_send_buffs;
- std::list<udp_zero_copy_asio_msb> _msb_pool;
- std::list<udp_zero_copy_asio_mrb> _mrb_pool;
+ std::vector<boost::shared_ptr<udp_zero_copy_asio_msb> > _msb_pool;
+ std::vector<boost::shared_ptr<udp_zero_copy_asio_mrb> > _mrb_pool;
+ size_t _next_recv_buff_index, _next_send_buff_index;
//asio guts -> socket and service
asio::io_service _io_service;
diff --git a/host/lib/transport/usb_zero_copy_wrapper.cpp b/host/lib/transport/usb_zero_copy_wrapper.cpp
index 3571ed856..87e001fed 100644
--- a/host/lib/transport/usb_zero_copy_wrapper.cpp
+++ b/host/lib/transport/usb_zero_copy_wrapper.cpp
@@ -15,12 +15,13 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//
+#include "simple_claimer.hpp"
#include <uhd/transport/usb_zero_copy.hpp>
-#include <uhd/transport/bounded_buffer.hpp>
#include <uhd/transport/buffer_pool.hpp>
#include <uhd/utils/byteswap.hpp>
#include <uhd/utils/msg.hpp>
#include <boost/foreach.hpp>
+#include <boost/make_shared.hpp>
#include <vector>
#include <iostream>
@@ -31,30 +32,39 @@ using namespace uhd::transport;
**********************************************************************/
class usb_zero_copy_wrapper_mrb : public managed_recv_buffer{
public:
- usb_zero_copy_wrapper_mrb(bounded_buffer<usb_zero_copy_wrapper_mrb *> &queue):
- _queue(queue){/*NOP*/}
+ usb_zero_copy_wrapper_mrb(void){/*NOP*/}
void release(void){
- if (not _mrb) return;
_mrb.reset(); //decrement ref count, other MRB's may hold a ref
- _queue.push_with_haste(this);
+ _claimer.release();
}
- UHD_INLINE sptr get_new(managed_recv_buffer::sptr mrb, const void *mem, size_t len){
+ UHD_INLINE sptr get_new(
+ managed_recv_buffer::sptr &mrb, size_t &offset_bytes,
+ const double timeout, size_t &index
+ ){
+ if (not mrb or not _claimer.claim_with_wait(timeout)) return sptr();
+
+ index++; //advances the caller's buffer
+
+ //hold a copy of the buffer shared pointer
_mrb = mrb;
- _mem = mem;
- _len = len;
- return make_managed_buffer(this);
+
+ //extract this packet's memory address and length in bytes
+ char *mem = mrb->cast<char *>() + offset_bytes;
+ const boost::uint32_t *mem32 = reinterpret_cast<const boost::uint32_t *>(mem);
+ size_t len = (uhd::wtohx(mem32[0]) & 0xffff)*sizeof(boost::uint32_t); //length in bytes (from VRT header)
+
+ //check if this receive buffer has been exhausted
+ offset_bytes += len;
+ if (offset_bytes >= mrb->size()) mrb.reset(); //drop caller's ref
+
+ return make(this, mem, len);
}
private:
- const void *get_buff(void) const{return _mem;}
- size_t get_size(void) const{return _len;}
-
- bounded_buffer<usb_zero_copy_wrapper_mrb *> &_queue;
- const void *_mem;
- size_t _len;
managed_recv_buffer::sptr _mrb;
+ simple_claimer _claimer;
};
/***********************************************************************
@@ -65,14 +75,12 @@ public:
usb_zero_copy_wrapper_msb(const usb_zero_copy::sptr internal, const size_t fragmentation_size):
_internal(internal), _fragmentation_size(fragmentation_size){/*NOP*/}
- void commit(size_t len){
- if (len == 0) return;
-
+ void release(void){
//get a reference to the VITA header before incrementing
const boost::uint32_t vita_header = reinterpret_cast<const boost::uint32_t *>(_mem_buffer_tip)[0];
- _bytes_in_buffer += len;
- _mem_buffer_tip += len;
+ _bytes_in_buffer += size();
+ _mem_buffer_tip += size();
//extract VITA end of packet flag, we must force flush under eof conditions
const bool eop = (uhd::wtohx(vita_header) & (0x1 << 24)) != 0;
@@ -90,13 +98,10 @@ public:
_mem_buffer_tip = _last_send_buff->cast<char *>();
_bytes_in_buffer = 0;
}
- return make_managed_buffer(this);
+ return make(this, _mem_buffer_tip, _fragmentation_size);
}
private:
- void *get_buff(void) const{return reinterpret_cast<void *>(_mem_buffer_tip);}
- size_t get_size(void) const{return _fragmentation_size;}
-
usb_zero_copy::sptr _internal;
const size_t _fragmentation_size;
managed_send_buffer::sptr _last_send_buff;
@@ -112,44 +117,26 @@ public:
usb_zero_copy_wrapper(sptr usb_zc, const size_t frame_boundary):
_internal_zc(usb_zc),
_frame_boundary(frame_boundary),
- _available_recv_buffs(this->get_num_recv_frames()),
- _mrb_pool(this->get_num_recv_frames(), usb_zero_copy_wrapper_mrb(_available_recv_buffs)),
- _the_only_msb(usb_zero_copy_wrapper_msb(usb_zc, frame_boundary))
+ _next_recv_buff_index(0)
{
- BOOST_FOREACH(usb_zero_copy_wrapper_mrb &mrb, _mrb_pool){
- _available_recv_buffs.push_with_haste(&mrb);
+ for (size_t i = 0; i < this->get_num_recv_frames(); i++){
+ _mrb_pool.push_back(boost::make_shared<usb_zero_copy_wrapper_mrb>());
}
+ _the_only_msb = boost::make_shared<usb_zero_copy_wrapper_msb>(usb_zc, frame_boundary);
}
managed_recv_buffer::sptr get_recv_buff(double timeout){
//attempt to get a managed recv buffer
- if (not _last_recv_buff.get()){
+ if (not _last_recv_buff){
_last_recv_buff = _internal_zc->get_recv_buff(timeout);
- _last_recv_offset = 0;
+ _last_recv_offset = 0; //reset offset into buffer
}
- //attempt to get a wrapper for a managed recv buffer
- usb_zero_copy_wrapper_mrb *wmrb = NULL;
- if (_last_recv_buff.get() and _available_recv_buffs.pop_with_timed_wait(wmrb, timeout)){
- //extract this packet's memory address and length in bytes
- const char *mem = _last_recv_buff->cast<const char *>() + _last_recv_offset;
- const boost::uint32_t *mem32 = reinterpret_cast<const boost::uint32_t *>(mem);
- const size_t len = (uhd::wtohx(mem32[0]) & 0xffff)*sizeof(boost::uint32_t); //length in bytes (from VRT header)
-
- managed_recv_buffer::sptr recv_buff; //the buffer to be returned to the user
- recv_buff = wmrb->get_new(_last_recv_buff, mem, len);
- _last_recv_offset += len;
-
- //check if this receive buffer has been exhausted
- if (_last_recv_offset >= _last_recv_buff->size()) {
- _last_recv_buff.reset();
- }
-
- return recv_buff;
- }
-
- //otherwise return a null sptr for failure
- return managed_recv_buffer::sptr();
+ //get the buffer to be returned to the user
+ if (_next_recv_buff_index == _mrb_pool.size()) _next_recv_buff_index = 0;
+ return _mrb_pool[_next_recv_buff_index]->get_new(
+ _last_recv_buff, _last_recv_offset, timeout, _next_recv_buff_index
+ );
}
size_t get_num_recv_frames(void) const{
@@ -161,7 +148,7 @@ public:
}
managed_send_buffer::sptr get_send_buff(double timeout){
- return _the_only_msb.get_new(timeout);
+ return _the_only_msb->get_new(timeout);
}
size_t get_num_send_frames(void) const{
@@ -175,16 +162,13 @@ public:
private:
sptr _internal_zc;
size_t _frame_boundary;
- bounded_buffer<usb_zero_copy_wrapper_mrb *> _available_recv_buffs;
- std::vector<usb_zero_copy_wrapper_mrb> _mrb_pool;
- usb_zero_copy_wrapper_msb _the_only_msb;
-
- //buffer to store partially-received VRT packets in
- buffer_pool::sptr _fragment_mem;
+ std::vector<boost::shared_ptr<usb_zero_copy_wrapper_mrb> > _mrb_pool;
+ boost::shared_ptr<usb_zero_copy_wrapper_msb> _the_only_msb;
//state for last recv buffer to create multiple managed buffers
managed_recv_buffer::sptr _last_recv_buff;
size_t _last_recv_offset;
+ size_t _next_recv_buff_index;
};
/***********************************************************************
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-17 10:55:17 +0000