7 files changed, 321 insertions, 629 deletions

diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt
index 8765c6703..a5bf9c5f1 100644
--- a/host/lib/transport/CMakeLists.txt
+++ b/host/lib/transport/CMakeLists.txt
@@ -48,20 +48,36 @@ ENDIF(ENABLE_USB)
 ########################################################################
 MESSAGE(STATUS "")
 MESSAGE(STATUS "Configuring interface address discovery...")
-
+INCLUDE(CheckCXXSourceCompiles)
 INCLUDE(CheckIncludeFileCXX)
-CHECK_INCLUDE_FILE_CXX(ifaddrs.h HAVE_IFADDRS_H)
+
+CHECK_CXX_SOURCE_COMPILES("
+    #include <ifaddrs.h>
+    int main(){
+        struct ifaddrs *ifap;
+        getifaddrs(&ifap);
+        return 0;
+    }
+    " HAVE_GETIFADDRS
+)
+
 CHECK_INCLUDE_FILE_CXX(winsock2.h HAVE_WINSOCK2_H)
 
-IF(HAVE_IFADDRS_H)
+IF(HAVE_GETIFADDRS)
     MESSAGE(STATUS "  Interface address discovery supported through getifaddrs.")
-    ADD_DEFINITIONS(-DHAVE_IFADDRS_H)
+    SET(IF_ADDRS_DEFS HAVE_GETIFADDRS)
 ELSEIF(HAVE_WINSOCK2_H)
     MESSAGE(STATUS "  Interface address discovery supported through SIO_GET_INTERFACE_LIST.")
-    ADD_DEFINITIONS(-DHAVE_WINSOCK2_H)
-ELSE(HAVE_IFADDRS_H)
+    SET(IF_ADDRS_DEFS HAVE_SIO_GET_INTERFACE_LIST)
+ELSE()
     MESSAGE(STATUS "  Interface address discovery not supported.")
-ENDIF(HAVE_IFADDRS_H)
+    SET(IF_ADDRS_DEFS HAVE_IF_ADDRS_DUMMY)
+ENDIF()
+
+SET_SOURCE_FILES_PROPERTIES(
+    ${CMAKE_CURRENT_SOURCE_DIR}/if_addrs.cpp
+    PROPERTIES COMPILE_DEFINITIONS "${IF_ADDRS_DEFS}"
+)
 
 ########################################################################
 # Append to the list of sources for lib uhd
@@ -75,7 +91,6 @@ LIBUHD_APPEND_SOURCES(
     ${CMAKE_CURRENT_SOURCE_DIR}/buffer_pool.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/if_addrs.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/udp_simple.cpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/udp_zero_copy_asio.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/udp_zero_copy.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/vrt_packet_handler.hpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/zero_copy.cpp
 )
diff --git a/host/lib/transport/if_addrs.cpp b/host/lib/transport/if_addrs.cpp
index 17cf8455b..b7c8ad844 100644
--- a/host/lib/transport/if_addrs.cpp
+++ b/host/lib/transport/if_addrs.cpp
@@ -20,14 +20,10 @@
 #include <boost/cstdint.hpp>
 #include <iostream>
 
-uhd::transport::if_addrs_t::if_addrs_t(void){
-    /* NOP */
-}
-
 /***********************************************************************
  * Interface address discovery through ifaddrs api
  **********************************************************************/
-#if defined(HAVE_IFADDRS_H)
+#ifdef HAVE_GETIFADDRS
 #include <ifaddrs.h>
 
 static boost::asio::ip::address_v4 sockaddr_to_ip_addr(sockaddr *addr){
@@ -59,10 +55,12 @@ std::vector<uhd::transport::if_addrs_t> uhd::transport::get_if_addrs(void){
     return if_addrs;
 }
 
+#endif /* HAVE_GETIFADDRS */
+
 /***********************************************************************
  * Interface address discovery through windows api
  **********************************************************************/
-#elif defined(HAVE_WINSOCK2_H)
+#ifdef HAVE_SIO_GET_INTERFACE_LIST
 #include <winsock2.h>
 
 std::vector<uhd::transport::if_addrs_t> uhd::transport::get_if_addrs(void){
@@ -98,13 +96,15 @@ std::vector<uhd::transport::if_addrs_t> uhd::transport::get_if_addrs(void){
     return if_addrs;
 }
 
+#endif /* HAVE_SIO_GET_INTERFACE_LIST */
+
 /***********************************************************************
  * Interface address discovery not included
  **********************************************************************/
-#else /* HAVE_IFADDRS_H */
+#ifdef HAVE_IF_ADDRS_DUMMY
 
 std::vector<uhd::transport::if_addrs_t> uhd::transport::get_if_addrs(void){
     return std::vector<if_addrs_t>();
 }
 
-#endif /* HAVE_IFADDRS_H */
+#endif /* HAVE_IF_ADDRS_DUMMY */
diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index 6fab5ae6f..87adece45 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -21,258 +21,88 @@
 #include <uhd/transport/buffer_pool.hpp>
 #include <uhd/utils/thread_priority.hpp>
 #include <uhd/utils/assert.hpp>
+#include <boost/function.hpp>
 #include <boost/foreach.hpp>
-#include <boost/thread.hpp>
-#include <vector>
+#include <boost/thread/thread.hpp>
+#include <list>
 #include <iostream>
 
 using namespace uhd;
 using namespace uhd::transport;
 
-static const double CLEANUP_TIMEOUT   = 0.2;    //seconds
 static const size_t DEFAULT_NUM_XFERS = 16;     //num xfers
 static const size_t DEFAULT_XFER_SIZE = 32*512; //bytes
 
 /***********************************************************************
- * Helper functions
- ***********************************************************************/
-/*
- * Print the values of a libusb_transfer struct
- * http://libusb.sourceforge.net/api-1.0/structlibusb__transfer.html
- */
-void pp_transfer(libusb_transfer *lut)
-{
-    std::cout << "Libusb transfer"       << std::endl;
-    std::cout << "    flags:         0x" << std::hex << (unsigned int) lut->flags << std::endl;
-    std::cout << "    endpoint:      0x" << std::hex << (unsigned int) lut->endpoint << std::endl;
-    std::cout << "    type:          0x" << std::hex << (unsigned int) lut->type << std::endl;
-    std::cout << "    timeout:       "   << std::dec << lut->timeout << std::endl;
-    std::cout << "    status:        0x" << std::hex << lut->status << std::endl;
-    std::cout << "    length:        "   << std::dec << lut->length << std::endl;
-    std::cout << "    actual_length: "   << std::dec << lut->actual_length << std::endl;
-}
-
-/***********************************************************************
- * USB asynchronous zero_copy endpoint
- *   This endpoint implementation provides asynchronous I/O to libusb-1.0
- *   devices. Each endpoint is directional and two can be combined to
- *   create a bidirectional interface. It is a zero copy implementation
- *   with respect to libusb, however, each send and recv requires a copy
- *   operation from kernel to userspace; this is due to the usbfs
- *   interface provided by the kernel.
+ * Reusable managed receiver buffer:
+ *  - Associated with a particular libusb transfer struct.
+ *  - Submits the transfer to libusb in the release method.
  **********************************************************************/
-class usb_endpoint {
+class libusb_zero_copy_mrb : public managed_recv_buffer{
 public:
-    typedef boost::shared_ptr<usb_endpoint> sptr;
-
-    usb_endpoint(
-        libusb::device_handle::sptr handle,
-        int endpoint,
-        bool input,
-        size_t transfer_size,
-        size_t num_transfers
-    );
-
-    ~usb_endpoint(void);
+    libusb_zero_copy_mrb(libusb_transfer *lut):
+        _lut(lut), _expired(true) { /* NOP */ }
 
-    // Exposed interface for submitting / retrieving transfer buffers
-
-    //! Submit a new transfer that was presumably just filled or emptied.
-    void submit(libusb_transfer *lut);
-
-    /*!
-     * Get an available transfer:
-     * For inputs, this is a just filled transfer.
-     * For outputs, this is a just emptied transfer.
-     * \param timeout the timeout to wait for a lut
-     * \return the transfer pointer or NULL if timeout
-     */
-    libusb_transfer *get_lut_with_wait(double timeout);
+    void release(void){
+        if (_expired) return;
+        UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
+        _expired = true;
+    }
 
-    //Callback use only
-    void callback_handle_transfer(libusb_transfer *lut);
+    sptr get_new(void){
+        _expired = false;
+        return sptr(this, &libusb_zero_copy_mrb::fake_deleter);
+    }
 
 private:
-    libusb::device_handle::sptr _handle;
-    int  _endpoint;
-    bool _input;
-
-    //! hold a bounded buffer of completed transfers
-    bounded_buffer<libusb_transfer *> _completed_list;
-
-    //! a list of all transfer structs we allocated
-    std::vector<libusb_transfer *> _all_luts;
+    static void fake_deleter(void *obj){
+        static_cast<libusb_zero_copy_mrb *>(obj)->release();
+    }
 
-    //! memory allocated for the transfer buffers
-    buffer_pool::sptr _buffer_pool;
+    const void *get_buff(void) const{return _lut->buffer;}
+    size_t get_size(void) const{return _lut->actual_length;}
 
-    // Calls for processing asynchronous I/O
-    libusb_transfer *allocate_transfer(void *mem, size_t len);
-    void print_transfer_status(libusb_transfer *lut);
+    libusb_transfer *_lut;
+    bool _expired;
 };
 
-
-/*
- * Callback function called when submitted transfers complete.
- * The endpoint upon which the transfer is part of is recovered
- * and the transfer moved from pending to completed state.
- * Callbacks occur during the reaping calls where libusb_handle_events()
- * is used. The callback only modifies the transfer state by moving
- * it from the pending to completed status list.
- * \param lut pointer to libusb_transfer
- */
-static void callback(libusb_transfer *lut){
-    usb_endpoint *endpoint = (usb_endpoint *) lut->user_data;
-    endpoint->callback_handle_transfer(lut);
-}
-
-
-/*
- * Accessor call to allow list access from callback space
- * \param pointer to libusb_transfer
- */
-void usb_endpoint::callback_handle_transfer(libusb_transfer *lut){
-    _completed_list.push_with_haste(lut);
-}
-
-
-/*
- * Constructor
- * Allocate libusb transfers and mark as free.  For IN endpoints,
- * submit the transfers so that they're ready to return when
- * data is available.
- */
-usb_endpoint::usb_endpoint(
-    libusb::device_handle::sptr handle,
-    int endpoint,
-    bool input,
-    size_t transfer_size,
-    size_t num_transfers
-):
-    _handle(handle),
-    _endpoint(endpoint),
-    _input(input),
-    _completed_list(num_transfers)
-{
-    _buffer_pool = buffer_pool::make(num_transfers, transfer_size);
-    for (size_t i = 0; i < num_transfers; i++){
-        _all_luts.push_back(allocate_transfer(_buffer_pool->at(i), transfer_size));
-
-        //input luts are immediately submitted to be filled
-        //output luts go into the completed list as free buffers
-        if (_input) this->submit(_all_luts.back());
-        else _completed_list.push_with_haste(_all_luts.back());
+/***********************************************************************
+ * 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;
+        UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
+        _expired = true;
     }
-}
-
 
-/*
- * Destructor
- * Make sure all the memory is freed. Cancel any pending transfers.
- * When all completed transfers are moved to the free list, release
- * the transfers. Libusb will deallocate the data buffer held by
- * each transfer.
- */
-usb_endpoint::~usb_endpoint(void){
-    //cancel all transfers
-    BOOST_FOREACH(libusb_transfer *lut, _all_luts){
-        libusb_cancel_transfer(lut);
+    sptr get_new(void){
+        _expired = false;
+        return sptr(this, &libusb_zero_copy_msb::fake_deleter);
     }
 
-    //collect canceled transfers (drain the queue)
-    while (this->get_lut_with_wait(CLEANUP_TIMEOUT) != NULL){};
-
-    //free all transfers
-    BOOST_FOREACH(libusb_transfer *lut, _all_luts){
-        libusb_free_transfer(lut);
+private:
+    static void fake_deleter(void *obj){
+        static_cast<libusb_zero_copy_msb *>(obj)->commit(0);
     }
-}
-
-
-/*
- * Allocate a libusb transfer
- * The allocated transfer - and buffer it contains - is repeatedly
- * submitted, reaped, and reused and should not be freed until shutdown.
- * \param mem a pointer to the buffer memory
- * \param len size of the individual buffer
- * \return pointer to an allocated libusb_transfer
- */
-libusb_transfer *usb_endpoint::allocate_transfer(void *mem, size_t len){
-    libusb_transfer *lut = libusb_alloc_transfer(0);
-    UHD_ASSERT_THROW(lut != NULL);
-
-    unsigned int endpoint = ((_endpoint & 0x7f) | (_input ? 0x80 : 0));
-    unsigned char *buff = reinterpret_cast<unsigned char *>(mem);
-    libusb_transfer_cb_fn lut_callback = libusb_transfer_cb_fn(&callback);
-
-    libusb_fill_bulk_transfer(lut,                // transfer
-                              _handle->get(),     // dev_handle
-                              endpoint,           // endpoint
-                              buff,               // buffer
-                              len,                // length
-                              lut_callback,       // callback
-                              this,               // user_data
-                              0);                 // timeout
-    return lut;
-}
 
+    void *get_buff(void) const{return _lut->buffer;}
+    size_t get_size(void) const{return _lut->length;}
 
-/*
- * Asynchonous transfer submission
- * Submit a libusb transfer to libusb add pending status
- * \param lut pointer to libusb_transfer
- * \return true on success or false on error
- */
-void usb_endpoint::submit(libusb_transfer *lut){
-    UHD_ASSERT_THROW(libusb_submit_transfer(lut) == 0);
-}
-
-/*
- * Print status errors of a completed transfer
- * \param lut pointer to an libusb_transfer
- */
-void usb_endpoint::print_transfer_status(libusb_transfer *lut){
-    std::cout << "here " << lut->status << std::endl;
-    switch (lut->status) {
-    case LIBUSB_TRANSFER_COMPLETED:
-        if (lut->actual_length < lut->length) {
-            std::cerr << "USB: transfer completed with short write,"
-                      << " length = " << lut->length
-                      << " actual = " << lut->actual_length << std::endl;
-        }
-
-        if ((lut->actual_length < 0) || (lut->length < 0)) {
-            std::cerr << "USB: transfer completed with invalid response"
-                      << std::endl;
-        }
-        break;
-    case LIBUSB_TRANSFER_CANCELLED:
-        break;
-    case LIBUSB_TRANSFER_NO_DEVICE:
-        std::cerr << "USB: device was disconnected" << std::endl;
-        break;
-    case LIBUSB_TRANSFER_OVERFLOW:
-        std::cerr << "USB: device sent more data than requested" << std::endl;
-        break;
-    case LIBUSB_TRANSFER_TIMED_OUT:
-        std::cerr << "USB: transfer timed out" << std::endl;
-        break;
-    case LIBUSB_TRANSFER_STALL:
-        std::cerr << "USB: halt condition detected (stalled)" << std::endl;
-        break;
-    case LIBUSB_TRANSFER_ERROR:
-        std::cerr << "USB: transfer failed" << std::endl;
-        break;
-    default:
-        std::cerr << "USB: received unknown transfer status" << std::endl;
-    }
-}
+    libusb_transfer *_lut;
+    bool _expired;
+};
 
-libusb_transfer *usb_endpoint::get_lut_with_wait(double timeout){
-    boost::this_thread::disable_interruption di; //disable because the wait can throw
-    libusb_transfer *lut = NULL;
-    if (_completed_list.pop_with_timed_wait(lut, timeout)) return lut;
-    return NULL;
+//! helper function: handles all async callbacks
+static void libusb_async_cb(libusb_transfer *lut){
+    (*static_cast<boost::function<void()> *>(lut->user_data))();
 }
 
 /***********************************************************************
@@ -286,16 +116,107 @@ public:
         size_t recv_endpoint,
         size_t send_endpoint,
         const device_addr_t &hints
-    );
+    ):
+        _handle(handle),
+        _recv_frame_size(size_t(hints.cast<double>("recv_frame_size", DEFAULT_XFER_SIZE))),
+        _num_recv_frames(size_t(hints.cast<double>("num_recv_frames", DEFAULT_NUM_XFERS))),
+        _send_frame_size(size_t(hints.cast<double>("send_frame_size", DEFAULT_XFER_SIZE))),
+        _num_send_frames(size_t(hints.cast<double>("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<unsigned char *>(_recv_buffer_pool->at(i)), // buffer
+                this->get_recv_frame_size(),                            // length
+                libusb_transfer_cb_fn(&libusb_async_cb),                // callback
+                static_cast<void *>(&_callbacks.back()),                // user_data
+                0                                                       // timeout
+            );
+
+            _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<unsigned char *>(_send_buffer_pool->at(i)), // buffer
+                this->get_send_frame_size(),                            // length
+                libusb_transfer_cb_fn(&libusb_async_cb),                // callback
+                static_cast<void *>(&_callbacks.back()),                // user_data
+                0                                                       // timeout
+            );
+
+            _all_luts.push_back(lut);
+            libusb_async_cb(lut);
+        }
+
+        //spawn the event handler threads
+        size_t concurrency = hints.cast<size_t>("concurrency_hint", 1);
+        for (size_t i = 0; i < concurrency; i++) _thread_group.create_thread(
+            boost::bind(&libusb_zero_copy_impl::run_event_loop, this)
+        );
+    }
 
     ~libusb_zero_copy_impl(void){
+        //shutdown the threads
         _threads_running = false;
         _thread_group.interrupt_all();
         _thread_group.join_all();
+
+        //cancel and free all transfers
+        BOOST_FOREACH(libusb_transfer *lut, _all_luts){
+            libusb_cancel_transfer(lut);
+            libusb_free_transfer(lut);
+        }
     }
 
-    managed_recv_buffer::sptr get_recv_buff(double);
-    managed_send_buffer::sptr get_send_buff(double);
+    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; }
@@ -304,125 +225,50 @@ public:
     size_t get_send_frame_size(void) const { return _send_frame_size; }
 
 private:
-    void release(libusb_transfer *lut){
-        _recv_ep->submit(lut);
+    //! Handle a bound async callback for recv
+    void handle_recv(libusb_zero_copy_mrb *mrb){
+        _pending_recv_buffs.push_with_haste(mrb);
     }
 
-    void commit(libusb_transfer *lut, size_t num_bytes){
-        lut->length = num_bytes;
-        try{
-            _send_ep->submit(lut);
-        }
-        catch(const std::exception &e){
-            std::cerr << "Error in commit: " << e.what() << std::endl;
-        }
+    //! 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;
-    usb_endpoint::sptr _recv_ep, _send_ep;
 
-    //event handler threads
+    //! Storage for transfer related objects
+    buffer_pool::sptr _recv_buffer_pool, _send_buffer_pool;
+    bounded_buffer<libusb_zero_copy_mrb *> _pending_recv_buffs;
+    bounded_buffer<libusb_zero_copy_msb *> _pending_send_buffs;
+    std::list<libusb_zero_copy_mrb> _mrb_pool;
+    std::list<libusb_zero_copy_msb> _msb_pool;
+    std::list<boost::function<void()> > _callbacks;
+
+    //! a list of all transfer structs we allocated
+    std::list<libusb_transfer *> _all_luts;
+
+    //! event handler threads
     boost::thread_group _thread_group;
     bool _threads_running;
 
     void run_event_loop(void){
         set_thread_priority_safe();
-        libusb::session::sptr session = libusb::session::get_global_session();
+        libusb_context *context = libusb::session::get_global_session()->get_context();
         _threads_running = true;
         try{
             while(_threads_running){
                 timeval tv;
                 tv.tv_sec = 0;
                 tv.tv_usec = 100000; //100ms
-                libusb_handle_events_timeout(session->get_context(), &tv);
+                libusb_handle_events_timeout(context, &tv);
             }
         } catch(const boost::thread_interrupted &){}
     }
-};
-
-/*
- * Constructor
- * Initializes libusb, opens devices, and sets up interfaces for I/O.
- * Finally, creates endpoints for asynchronous I/O.
- */
-libusb_zero_copy_impl::libusb_zero_copy_impl(
-    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<double>("recv_frame_size", DEFAULT_XFER_SIZE))),
-    _num_recv_frames(size_t(hints.cast<double>("num_recv_frames", DEFAULT_NUM_XFERS))),
-    _send_frame_size(size_t(hints.cast<double>("send_frame_size", DEFAULT_XFER_SIZE))),
-    _num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_XFERS)))
-{
-    _handle->claim_interface(2 /*in interface*/);
-    _handle->claim_interface(1 /*out interface*/);
-
-    _recv_ep = usb_endpoint::sptr(new usb_endpoint(
-                              _handle,         // libusb device_handle
-                              recv_endpoint,   // USB endpoint number
-                              true,            // IN endpoint
-                              this->get_recv_frame_size(),  // buffer size per transfer
-                              this->get_num_recv_frames()   // number of libusb transfers
-    ));
-
-    _send_ep = usb_endpoint::sptr(new usb_endpoint(
-                              _handle,         // libusb device_handle
-                              send_endpoint,   // USB endpoint number
-                              false,           // OUT endpoint
-                              this->get_send_frame_size(),  // buffer size per transfer
-                              this->get_num_send_frames()   // number of libusb transfers
-    ));
-
-    //spawn the event handler threads
-    size_t concurrency = hints.cast<size_t>("concurrency_hint", 1);
-    for (size_t i = 0; i < concurrency; i++) _thread_group.create_thread(
-        boost::bind(&libusb_zero_copy_impl::run_event_loop, this)
-    );
-}
-
-/*
- * Construct a managed receive buffer from a completed libusb transfer
- * (happy with buffer full of data) obtained from the receive endpoint.
- * Return empty pointer if no transfer is available (timeout or error).
- * \return pointer to a managed receive buffer
- */
-managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(double timeout){
-    libusb_transfer *lut = _recv_ep->get_lut_with_wait(timeout);
-    if (lut == NULL) {
-        return managed_recv_buffer::sptr();
-    }
-    else {
-        return managed_recv_buffer::make_safe(
-            lut->buffer, lut->actual_length,
-            boost::bind(&libusb_zero_copy_impl::release, this, lut)
-        );
-    }
-}
 
-
-/*
- * Construct a managed send buffer from a free libusb transfer (with
- * empty buffer). Return empty pointer of no transfer is available
- * (timeout or error).
- * \return pointer to a managed send buffer
- */
-managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){
-    libusb_transfer *lut = _send_ep->get_lut_with_wait(timeout);
-    if (lut == NULL) {
-        return managed_send_buffer::sptr();
-    }
-    else {
-        return managed_send_buffer::make_safe(
-            lut->buffer, this->get_send_frame_size(),
-            boost::bind(&libusb_zero_copy_impl::commit, this, lut, _1)
-        );
-    }
-}
+};
 
 /***********************************************************************
  * USB zero_copy make functions
diff --git a/host/lib/transport/udp_common.hpp b/host/lib/transport/udp_common.hpp
new file mode 100644
index 000000000..47775d9c4
--- /dev/null
+++ b/host/lib/transport/udp_common.hpp
@@ -0,0 +1,53 @@
+//
+// Copyright 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/>.
+//
+
+#ifndef INCLUDED_LIBUHD_TRANSPORT_VRT_PACKET_HANDLER_HPP
+#define INCLUDED_LIBUHD_TRANSPORT_VRT_PACKET_HANDLER_HPP
+
+#include <uhd/config.hpp>
+#include <boost/asio.hpp>
+
+namespace uhd{ namespace transport{
+
+    typedef boost::shared_ptr<boost::asio::ip::udp::socket> socket_sptr;
+
+    /*!
+     * Wait for the socket to become ready for a receive operation.
+     * \param sock_fd the open socket file descriptor
+     * \param timeout the timeout duration in seconds
+     * \return true when the socket is ready for receive
+     */
+    UHD_INLINE bool wait_for_recv_ready(int sock_fd, double timeout){
+        //setup timeval for timeout
+        timeval tv;
+        //If the tv_usec > 1 second on some platforms, select will
+        //error EINVAL: An invalid timeout interval was specified.
+        tv.tv_sec = int(timeout);
+        tv.tv_usec = int(timeout*1000000)%1000000;
+
+        //setup rset for timeout
+        fd_set rset;
+        FD_ZERO(&rset);
+        FD_SET(sock_fd, &rset);
+
+        //call select with timeout on receive socket
+        return ::select(sock_fd+1, &rset, NULL, NULL, &tv) > 0;
+    }
+
+}} //namespace uhd::transport
+
+#endif /* INCLUDED_LIBUHD_TRANSPORT_VRT_PACKET_HANDLER_HPP */
diff --git a/host/lib/transport/udp_simple.cpp b/host/lib/transport/udp_simple.cpp
index 6799ac7b2..1ee036d52 100644
--- a/host/lib/transport/udp_simple.cpp
+++ b/host/lib/transport/udp_simple.cpp
@@ -1,5 +1,5 @@
 //
-// Copyright 2010 Ettus Research LLC
+// 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
@@ -15,159 +15,69 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 //
 
+#include "udp_common.hpp"
 #include <uhd/transport/udp_simple.hpp>
-#include <boost/asio.hpp>
-#include <boost/thread.hpp>
 #include <boost/format.hpp>
 #include <iostream>
 
 using namespace uhd::transport;
+namespace asio = boost::asio;
 
 /***********************************************************************
- * Helper Functions
+ * UDP simple implementation: connected and broadcast
  **********************************************************************/
-/*!
- * Wait for available data or timeout.
- * \param socket the asio socket
- * \param timeout the timeout in seconds
- * \return false for timeout, true for data
- */
-static bool wait_available(
-    boost::asio::ip::udp::socket &socket, double timeout
-){
-    #if defined(UHD_PLATFORM_LINUX) || defined(UHD_PLATFORM_WIN32)
-
-    //setup timeval for timeout
-    timeval tv;
-    tv.tv_sec = 0;
-    tv.tv_usec = long(timeout*1e6);
-
-    //setup rset for timeout
-    fd_set rset;
-    FD_ZERO(&rset);
-    FD_SET(socket.native(), &rset);
-
-    return ::select(socket.native()+1, &rset, NULL, NULL, &tv) > 0;
-
-    #else /*defined(UHD_PLATFORM_LINUX) || defined(UHD_PLATFORM_WIN32)*/
-
-    //FIXME: why does select fail on macintosh?
-    for (size_t i = 0; i < size_t(timeout*1e3); i++){
-        if (socket.available()) return true;
-        boost::this_thread::sleep(boost::posix_time::milliseconds(1));
-    }
-    return false;
-
-    #endif /*defined(UHD_PLATFORM_LINUX) || defined(UHD_PLATFORM_WIN32)*/
-}
-
-/***********************************************************************
- * UDP connected implementation class
- **********************************************************************/
-class udp_connected_impl : public udp_simple{
+class udp_simple_impl : public udp_simple{
 public:
-    //structors
-    udp_connected_impl(const std::string &addr, const std::string &port);
-    ~udp_connected_impl(void);
-
-    //send/recv
-    size_t send(const boost::asio::const_buffer &);
-    size_t recv(const boost::asio::mutable_buffer &, double);
-
-private:
-    boost::asio::ip::udp::socket   *_socket;
-    boost::asio::io_service        _io_service;
-};
+    udp_simple_impl(
+        const std::string &addr, const std::string &port, bool bcast, bool connect
+    ):_connected(connect){
+        //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl;
 
-udp_connected_impl::udp_connected_impl(const std::string &addr, const std::string &port){
-    //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl;
+        //resolve the address
+        asio::ip::udp::resolver resolver(_io_service);
+        asio::ip::udp::resolver::query query(asio::ip::udp::v4(), addr, port);
+        _receiver_endpoint = *resolver.resolve(query);
 
-    // resolve the address
-    boost::asio::ip::udp::resolver resolver(_io_service);
-    boost::asio::ip::udp::resolver::query query(boost::asio::ip::udp::v4(), addr, port);
-    boost::asio::ip::udp::endpoint receiver_endpoint = *resolver.resolve(query);
+        //create and open the socket
+        _socket = socket_sptr(new asio::ip::udp::socket(_io_service));
+        _socket->open(asio::ip::udp::v4());
 
-    // Create, open, and connect the socket
-    _socket = new boost::asio::ip::udp::socket(_io_service);
-    _socket->open(boost::asio::ip::udp::v4());
-    _socket->connect(receiver_endpoint);
-}
+        //allow broadcasting
+        _socket->set_option(asio::socket_base::broadcast(bcast));
 
-udp_connected_impl::~udp_connected_impl(void){
-    delete _socket;
-}
-
-size_t udp_connected_impl::send(const boost::asio::const_buffer &buff){
-    return _socket->send(boost::asio::buffer(buff));
-}
+        //connect the socket
+        if (connect) _socket->connect(_receiver_endpoint);
 
-size_t udp_connected_impl::recv(const boost::asio::mutable_buffer &buff, double timeout){
-    if (not wait_available(*_socket, timeout)) return 0;
-    return _socket->receive(boost::asio::buffer(buff));
-}
+    }
 
-/***********************************************************************
- * UDP broadcast implementation class
- **********************************************************************/
-class udp_broadcast_impl : public udp_simple{
-public:
-    //structors
-    udp_broadcast_impl(const std::string &addr, const std::string &port);
-    ~udp_broadcast_impl(void);
+    size_t send(const asio::const_buffer &buff){
+        if (_connected) return _socket->send(asio::buffer(buff));
+        return _socket->send_to(asio::buffer(buff), _receiver_endpoint);
+    }
 
-    //send/recv
-    size_t send(const boost::asio::const_buffer &);
-    size_t recv(const boost::asio::mutable_buffer &, double);
+    size_t recv(const asio::mutable_buffer &buff, double timeout){
+        if (not wait_for_recv_ready(_socket->native(), timeout)) return 0;
+        return _socket->receive(asio::buffer(buff));
+    }
 
 private:
-    boost::asio::ip::udp::socket   *_socket;
-    boost::asio::ip::udp::endpoint _receiver_endpoint;
-    boost::asio::io_service        _io_service;
+    bool                    _connected;
+    asio::io_service        _io_service;
+    socket_sptr             _socket;
+    asio::ip::udp::endpoint _receiver_endpoint;
 };
 
-udp_broadcast_impl::udp_broadcast_impl(const std::string &addr, const std::string &port){
-    //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl;
-
-    // resolve the address
-    boost::asio::ip::udp::resolver resolver(_io_service);
-    boost::asio::ip::udp::resolver::query query(boost::asio::ip::udp::v4(), addr, port);
-    _receiver_endpoint = *resolver.resolve(query);
-
-    // Create and open the socket
-    _socket = new boost::asio::ip::udp::socket(_io_service);
-    _socket->open(boost::asio::ip::udp::v4());
-
-    // Allow broadcasting
-    boost::asio::socket_base::broadcast option(true);
-    _socket->set_option(option);
-
-}
-
-udp_broadcast_impl::~udp_broadcast_impl(void){
-    delete _socket;
-}
-
-size_t udp_broadcast_impl::send(const boost::asio::const_buffer &buff){
-    return _socket->send_to(boost::asio::buffer(buff), _receiver_endpoint);
-}
-
-size_t udp_broadcast_impl::recv(const boost::asio::mutable_buffer &buff, double timeout){
-    if (not wait_available(*_socket, timeout)) return 0;
-    boost::asio::ip::udp::endpoint sender_endpoint;
-    return _socket->receive_from(boost::asio::buffer(buff), sender_endpoint);
-}
-
 /***********************************************************************
  * UDP public make functions
  **********************************************************************/
 udp_simple::sptr udp_simple::make_connected(
     const std::string &addr, const std::string &port
 ){
-    return sptr(new udp_connected_impl(addr, port));
+    return sptr(new udp_simple_impl(addr, port, false, true /* no bcast, connect */));
 }
 
 udp_simple::sptr udp_simple::make_broadcast(
     const std::string &addr, const std::string &port
 ){
-    return sptr(new udp_broadcast_impl(addr, port));
+    return sptr(new udp_simple_impl(addr, port, true, false /* bcast, no connect */));
 }
diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy.cpp
index 2794d383c..dda3bb547 100644
--- a/host/lib/transport/udp_zero_copy_asio.cpp
+++ b/host/lib/transport/udp_zero_copy.cpp
@@ -15,16 +15,15 @@
 // 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/bounded_buffer.hpp>
 #include <uhd/transport/buffer_pool.hpp>
-#include <uhd/utils/assert.hpp>
 #include <uhd/utils/warning.hpp>
-#include <boost/asio.hpp>
 #include <boost/format.hpp>
 #include <iostream>
-#include <vector>
+#include <list>
 
 using namespace uhd;
 using namespace uhd::transport;
@@ -40,20 +39,18 @@ static const size_t DEFAULT_NUM_FRAMES = 32;
  **********************************************************************/
 class udp_zero_copy_asio_mrb : public managed_recv_buffer{
 public:
-    typedef boost::shared_ptr<udp_zero_copy_asio_mrb> sptr;
     typedef boost::function<void(udp_zero_copy_asio_mrb *)> release_cb_type;
 
     udp_zero_copy_asio_mrb(void *mem, const release_cb_type &release_cb):
-        _mem(mem), _release_cb(release_cb){/* NOP */}
+        _mem(mem), _len(0), _release_cb(release_cb){/* NOP */}
 
     void release(void){
-        if (_expired) return;
+        if (_len == 0) return;
         this->_release_cb(this);
-        _expired = true;
+        _len = 0;
     }
 
     sptr get_new(size_t len){
-        _expired = false;
         _len = len;
         return sptr(this, &udp_zero_copy_asio_mrb::fake_deleter);
     }
@@ -68,7 +65,6 @@ private:
     const void *get_buff(void) const{return _mem;}
     size_t get_size(void) const{return _len;}
 
-    bool _expired;
     void *_mem;
     size_t _len;
     release_cb_type _release_cb;
@@ -81,20 +77,18 @@ private:
  **********************************************************************/
 class udp_zero_copy_asio_msb : public managed_send_buffer{
 public:
-    typedef boost::shared_ptr<udp_zero_copy_asio_msb> sptr;
     typedef boost::function<void(udp_zero_copy_asio_msb *, size_t)> commit_cb_type;
 
     udp_zero_copy_asio_msb(void *mem, const commit_cb_type &commit_cb):
-        _mem(mem), _commit_cb(commit_cb){/* NOP */}
+        _mem(mem), _len(0), _commit_cb(commit_cb){/* NOP */}
 
     void commit(size_t len){
-        if (_expired) return;
+        if (_len == 0) return;
         this->_commit_cb(this, len);
-        _expired = true;
+        _len = 0;
     }
 
     sptr get_new(size_t len){
-        _expired = false;
         _len = len;
         return sptr(this, &udp_zero_copy_asio_msb::fake_deleter);
     }
@@ -107,7 +101,6 @@ private:
     void *get_buff(void) const{return _mem;}
     size_t get_size(void) const{return _len;}
 
-    bool _expired;
     void *_mem;
     size_t _len;
     commit_cb_type _commit_cb;
@@ -135,7 +128,8 @@ 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)
+        _pending_recv_buffs(_num_recv_frames),
+        _pending_send_buffs(_num_send_frames)
     {
         //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl;
 
@@ -145,34 +139,28 @@ public:
         asio::ip::udp::endpoint receiver_endpoint = *resolver.resolve(query);
 
         //create, open, and connect the socket
-        _socket = new asio::ip::udp::socket(_io_service);
+        _socket = socket_sptr(new asio::ip::udp::socket(_io_service));
         _socket->open(asio::ip::udp::v4());
         _socket->connect(receiver_endpoint);
         _sock_fd = _socket->native();
 
         //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::sptr(
-                new udp_zero_copy_asio_mrb(_recv_buffer_pool->at(i),
+            _mrb_pool.push_back(udp_zero_copy_asio_mrb(_recv_buffer_pool->at(i),
                 boost::bind(&udp_zero_copy_asio_impl::release, this, _1))
-            ));
-            handle_recv(_mrb_pool.back().get());
+            );
+            handle_recv(&_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::sptr(
-                new udp_zero_copy_asio_msb(_send_buffer_pool->at(i),
+            _msb_pool.push_back(udp_zero_copy_asio_msb(_send_buffer_pool->at(i),
                 boost::bind(&udp_zero_copy_asio_impl::commit, this, _1, _2))
-            ));
-            handle_send(_msb_pool.back().get());
+            );
+            handle_send(&_msb_pool.back());
         }
     }
 
-    ~udp_zero_copy_asio_impl(void){
-        delete _socket;
-    }
-
     //get size for internal socket buffer
     template <typename Opt> size_t get_buff_size(void) const{
         Opt option;
@@ -190,30 +178,24 @@ public:
     /*******************************************************************
      * Receive implementation:
      *
-     * Use select to perform a blocking receive with timeout.
+     * 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.
      ******************************************************************/
-    UHD_INLINE bool is_recv_ready(double timeout){
-        //setup timeval for timeout
-        timeval tv;
-        tv.tv_sec = 0;
-        tv.tv_usec = long(timeout*1e6);
-
-        //setup rset for timeout
-        fd_set rset;
-        FD_ZERO(&rset);
-        FD_SET(_sock_fd, &rset);
-
-        //call select with timeout on receive socket
-        return ::select(_sock_fd+1, &rset, NULL, NULL, &tv) > 0;
-    }
-
     managed_recv_buffer::sptr get_recv_buff(double timeout){
         udp_zero_copy_asio_mrb *mrb = NULL;
-        if (is_recv_ready(timeout) and _pending_recv_buffs.pop_with_timed_wait(mrb, timeout)){
-            return mrb->get_new(::recv(_sock_fd, mrb->cast<char *>(), _recv_frame_size, 0));
+        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)
+            );
         }
         return managed_recv_buffer::sptr();
     }
@@ -264,12 +246,12 @@ private:
     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::vector<udp_zero_copy_asio_msb::sptr> _msb_pool;
-    std::vector<udp_zero_copy_asio_mrb::sptr> _mrb_pool;
+    std::list<udp_zero_copy_asio_msb> _msb_pool;
+    std::list<udp_zero_copy_asio_mrb> _mrb_pool;
 
     //asio guts -> socket and service
     asio::io_service        _io_service;
-    asio::ip::udp::socket   *_socket;
+    socket_sptr             _socket;
     int                     _sock_fd;
 };
 
diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp
deleted file mode 100644
index b91eaae1d..000000000
--- a/host/lib/transport/zero_copy.cpp
+++ /dev/null
@@ -1,114 +0,0 @@
-//
-// 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 <uhd/transport/zero_copy.hpp>
-
-using namespace uhd::transport;
-
-/***********************************************************************
- * Safe managed receive buffer
- **********************************************************************/
-static void release_nop(void){
-    /* NOP */
-}
-
-class safe_managed_receive_buffer : public managed_recv_buffer{
-public:
-    safe_managed_receive_buffer(
-        const void *buff, size_t size, const release_fcn_t &release_fcn
-    ):
-        _buff(buff), _size(size), _release_fcn(release_fcn)
-    {
-        /* NOP */
-    }
-
-    ~safe_managed_receive_buffer(void){
-        _release_fcn();
-    }
-
-    void release(void){
-        release_fcn_t release_fcn = _release_fcn;
-        _release_fcn = &release_nop;
-        return release_fcn();
-    }
-
-private:
-    const void *get_buff(void) const{
-        return _buff;
-    }
-
-    size_t get_size(void) const{
-        return _size;
-    }
-
-    const void *_buff;
-    size_t _size;
-    release_fcn_t _release_fcn;
-};
-
-managed_recv_buffer::sptr managed_recv_buffer::make_safe(
-    const void *buff, size_t size, const release_fcn_t &release_fcn
-){
-    return sptr(new safe_managed_receive_buffer(buff, size, release_fcn));
-}
-
-/***********************************************************************
- * Safe managed send buffer
- **********************************************************************/
-static void commit_nop(size_t){
-    /* NOP */
-}
-
-class safe_managed_send_buffer : public managed_send_buffer{
-public:
-    safe_managed_send_buffer(
-        void *buff, size_t size, const commit_fcn_t &commit_fcn
-    ):
-        _buff(buff), _size(size), _commit_fcn(commit_fcn)
-    {
-        /* NOP */
-    }
-
-    ~safe_managed_send_buffer(void){
-        _commit_fcn(0);
-    }
-
-    void commit(size_t num_bytes){
-        commit_fcn_t commit_fcn = _commit_fcn;
-        _commit_fcn = &commit_nop;
-        return commit_fcn(num_bytes);
-    }
-
-private:
-    void *get_buff(void) const{
-        return _buff;
-    }
-
-    size_t get_size(void) const{
-        return _size;
-    }
-
-    void *_buff;
-    size_t _size;
-    commit_fcn_t _commit_fcn;
-};
-
-safe_managed_send_buffer::sptr managed_send_buffer::make_safe(
-    void *buff, size_t size, const commit_fcn_t &commit_fcn
-){
-    return sptr(new safe_managed_send_buffer(buff, size, commit_fcn));
-}