uhd: squashed transport work

6 files changed, 492 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/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;
 };
 
 /***********************************************************************