uhd: reworked the zero copy interface

- recv buffers have a release call
- safe make function for recv buffers (buff + release callback)
- send buffer commits now have a void return
- safe make function for send buffers (buff + commit callback)

The reason for the void return from commit is that ssize_t num_bytes was never returning anything of use.
That is for all of the zero copy implementations so far, commit cannot really error (being asynchronous).

libusb zero copy impl was reworked to support the new interface.
USRP1 io_impl with the psuedo managed buffer was replaced with safe_make.
Also, usrp1 io_impl was simplified greatly due to commit returning void now.
UDP zero copy asio was disabled (in this commit, until its reworked).
Phony send and recv interfaces were removed completely.

6 files changed, 130 insertions, 325 deletions

diff --git a/host/include/uhd/transport/zero_copy.hpp b/host/include/uhd/transport/zero_copy.hpp
index ba19b193c..9dd16280c 100644
--- a/host/include/uhd/transport/zero_copy.hpp
+++ b/host/include/uhd/transport/zero_copy.hpp
@@ -19,10 +19,10 @@
 #define INCLUDED_UHD_TRANSPORT_ZERO_COPY_HPP
 
 #include <uhd/config.hpp>
-#include <uhd/utils/pimpl.hpp>
 #include <boost/asio/buffer.hpp>
 #include <boost/utility.hpp>
 #include <boost/shared_ptr.hpp>
+#include <boost/function.hpp>
 
 namespace uhd{ namespace transport{
 
@@ -34,14 +34,27 @@ namespace uhd{ namespace transport{
     class UHD_API managed_recv_buffer : boost::noncopyable{
     public:
         typedef boost::shared_ptr<managed_recv_buffer> sptr;
+        typedef boost::function<void(void)> release_fcn_t;
+
+        /*!
+         * Make a safe managed receive buffer:
+         * A safe managed buffer ensures that release is called once,
+         * either by the user or automatically upon deconstruction.
+         * \param buff a reference to the constant buffer
+         * \param release_fcn callback to release the memory
+         * \return a new managed receive buffer
+         */
+        static sptr make_safe(
+            const boost::asio::const_buffer &buff,
+            const release_fcn_t &release_fcn
+        );
 
         /*!
-         * Managed recv buffer destructor:
          * Signal to the transport that we are done with the buffer.
-         * This should be called to release the buffer to the transport.
+         * This should be called to release the buffer to the transport object.
          * After calling, the referenced memory should be considered invalid.
          */
-        virtual ~managed_recv_buffer(void) = 0;
+        virtual void release(void) = 0;
 
         /*!
          * Get the size of the underlying buffer.
@@ -71,20 +84,34 @@ namespace uhd{ namespace transport{
     /*!
      * A managed send buffer:
      * Contains a reference to transport-managed memory,
-     * and a method to release the memory after writing.
+     * and a method to commit the memory after writing.
      */
     class UHD_API managed_send_buffer : boost::noncopyable{
     public:
         typedef boost::shared_ptr<managed_send_buffer> sptr;
+        typedef boost::function<void(size_t)> commit_fcn_t;
+
+        /*!
+         * Make a safe managed send buffer:
+         * A safe managed buffer ensures that commit is called once,
+         * either by the user or automatically upon deconstruction.
+         * In the later case, the deconstructor will call commit(0).
+         * \param buff a reference to the mutable buffer
+         * \param commit_fcn callback to commit the memory
+         * \return a new managed send buffer
+         */
+        static sptr make_safe(
+            const boost::asio::mutable_buffer &buff,
+            const commit_fcn_t &commit_fcn
+        );
 
         /*!
          * Signal to the transport that we are done with the buffer.
          * This should be called to commit the write to the transport object.
          * After calling, the referenced memory should be considered invalid.
          * \param num_bytes the number of bytes written into the buffer
-         * \return the number of bytes written, 0 for timeout, negative for error
          */
-        virtual ssize_t commit(size_t num_bytes) = 0;
+        virtual void commit(size_t num_bytes) = 0;
 
         /*!
          * Get the size of the underlying buffer.
@@ -154,77 +181,6 @@ namespace uhd{ namespace transport{
 
     };
 
-    /*!
-     * A phony-zero-copy interface for transport objects that
-     * provides a zero-copy interface on top of copying transport.
-     * This interface implements the get managed recv buffer,
-     * the base class must implement the private recv method.
-     */
-    class UHD_API phony_zero_copy_recv_if : public virtual zero_copy_if{
-    public:
-        /*!
-         * Create a phony zero copy recv interface.
-         * \param max_buff_size max buffer size in bytes
-         */
-        phony_zero_copy_recv_if(size_t max_buff_size);
-
-        //! destructor
-        virtual ~phony_zero_copy_recv_if(void);
-
-        /*!
-         * Get a new receive buffer from this transport object.
-         * \param timeout the timeout to get the buffer in seconds
-         * \return a managed buffer, or null sptr on timeout/error
-         */
-        managed_recv_buffer::sptr get_recv_buff(double timeout);
-
-    private:
-        /*!
-         * Perform a private copying recv.
-         * \param buff the buffer to write data into
-         * \param timeout the timeout to get the buffer in seconds
-         * \return the number of bytes written to buff, 0 for timeout, negative for error
-         */
-        virtual ssize_t recv(const boost::asio::mutable_buffer &buff, double timeout) = 0;
-
-        UHD_PIMPL_DECL(impl) _impl;
-    };
-
-    /*!
-     * A phony-zero-copy interface for transport objects that
-     * provides a zero-copy interface on top of copying transport.
-     * This interface implements the get managed send buffer,
-     * the base class must implement the private send method.
-     */
-    class UHD_API phony_zero_copy_send_if : public virtual zero_copy_if{
-    public:
-        /*!
-         * Create a phony zero copy send interface.
-         * \param max_buff_size max buffer size in bytes
-         */
-        phony_zero_copy_send_if(size_t max_buff_size);
-
-        //! destructor
-        virtual ~phony_zero_copy_send_if(void);
-
-        /*!
-         * Get a new send buffer from this transport object.
-         * \param timeout the timeout to get the buffer in seconds
-         * \return a managed buffer, or null sptr on timeout/error
-         */
-        managed_send_buffer::sptr get_send_buff(double timeout);
-
-    private:
-        /*!
-         * Perform a private copying send.
-         * \param buff the buffer to read data from
-         * \return the number of bytes read from buff, 0 for timeout, negative for error
-         */
-        virtual ssize_t send(const boost::asio::const_buffer &buff) = 0;
-
-        UHD_PIMPL_DECL(impl) _impl;
-    };
-
 }} //namespace
 
 #endif /* INCLUDED_UHD_TRANSPORT_ZERO_COPY_HPP */
diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt
index 61616d077..2be2c89ec 100644
--- a/host/lib/transport/CMakeLists.txt
+++ b/host/lib/transport/CMakeLists.txt
@@ -36,6 +36,8 @@ IF(LIBUSB_FOUND)
         INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/lib/transport/msvc)
     ENDIF(MSVC)
     SET(HAVE_USB_SUPPORT TRUE)
+ELSE(LIBUSB_FOUND)
+    #TODO dummy usb
 ENDIF(LIBUSB_FOUND)
 
 IF(HAVE_USB_SUPPORT)
@@ -99,7 +101,7 @@ SET_SOURCE_FILES_PROPERTIES(
 LIBUHD_APPEND_SOURCES(
     ${CMAKE_SOURCE_DIR}/lib/transport/if_addrs.cpp
     ${CMAKE_SOURCE_DIR}/lib/transport/udp_simple.cpp
-    ${CMAKE_SOURCE_DIR}/lib/transport/udp_zero_copy_asio.cpp
+    #${CMAKE_SOURCE_DIR}/lib/transport/udp_zero_copy_asio.cpp
     ${CMAKE_SOURCE_DIR}/lib/transport/vrt_packet_handler.hpp
     ${CMAKE_SOURCE_DIR}/lib/transport/zero_copy.cpp
 )
diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index 7f2bc3468..e1cc8398c 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -95,9 +95,6 @@ private:
     int  _endpoint;
     bool _input;
 
-    size_t _transfer_size;
-    size_t _num_transfers;
-
     //! hold a bounded buffer of completed transfers
     typedef bounded_buffer<libusb_transfer *> lut_buff_type;
     lut_buff_type::sptr _completed_list;
@@ -154,14 +151,12 @@ usb_endpoint::usb_endpoint(
 ):
     _handle(handle),
     _endpoint(endpoint),
-    _input(input),
-    _transfer_size(transfer_size),
-    _num_transfers(num_transfers)
+    _input(input)
 {
     _completed_list = lut_buff_type::make(num_transfers);
 
-    for (size_t i = 0; i < _num_transfers; i++){
-        _all_luts.push_back(allocate_transfer(_transfer_size));
+    for (size_t i = 0; i < num_transfers; i++){
+        _all_luts.push_back(allocate_transfer(transfer_size));
 
         //input luts are immediately submitted to be filled
         //output luts go into the completed list as free buffers
@@ -280,126 +275,44 @@ libusb_transfer *usb_endpoint::get_lut_with_wait(double timeout){
 }
 
 /***********************************************************************
- * Managed buffers
+ * USB zero_copy device class
  **********************************************************************/
-/*
- * Libusb managed receive buffer
- * Construct a recv buffer from a libusb transfer. The memory held by
- * the libusb transfer is exposed through the managed buffer interface.
- * Upon destruction, the transfer and buffer are resubmitted to the
- * endpoint for further use.
- */
-class libusb_managed_recv_buffer_impl : public managed_recv_buffer {
+class libusb_zero_copy_impl : public usb_zero_copy {
 public:
-    libusb_managed_recv_buffer_impl(libusb_transfer *lut,
-                                    usb_endpoint::sptr endpoint)
-        : _buff(lut->buffer, lut->length)
-    {
-        _lut = lut;
-        _endpoint = endpoint;
-    }
-
-    ~libusb_managed_recv_buffer_impl(void){
-        _endpoint->submit(_lut);
-    }
+    typedef boost::shared_ptr<libusb_zero_copy_impl> sptr;
 
-private:
-    const boost::asio::const_buffer &get(void) const{
-        return _buff;
-    }
+    libusb_zero_copy_impl(
+        libusb::device_handle::sptr handle,
+        unsigned int recv_endpoint, unsigned int send_endpoint,
+        size_t recv_xfer_size, size_t recv_num_xfers,
+        size_t send_xfer_size, size_t send_num_xfers
+    );
 
-    libusb_transfer *_lut;
-    usb_endpoint::sptr _endpoint;
-    const boost::asio::const_buffer _buff;
-};
+    managed_recv_buffer::sptr get_recv_buff(double);
+    managed_send_buffer::sptr get_send_buff(double);
 
-/*
- * Libusb managed send buffer
- * Construct a send buffer from a libusb transfer. The memory held by
- * the libusb transfer is exposed through the managed buffer interface.
- * Committing the buffer will set the data length and submit the buffer
- * to the endpoint. Submitting a buffer multiple times or destroying
- * the buffer before committing is an error. For the latter, the transfer
- * is returned to the endpoint with no data for reuse.
- */
-class libusb_managed_send_buffer_impl : public managed_send_buffer {
-public:
-    libusb_managed_send_buffer_impl(libusb_transfer *lut,
-                                    usb_endpoint::sptr endpoint)
-        : _buff(lut->buffer, lut->length), _committed(false)
-    {
-        _lut = lut;
-        _endpoint = endpoint;
-    }
+    size_t get_num_recv_frames(void) const { return _recv_num_frames; }
+    size_t get_num_send_frames(void) const { return _send_num_frames; }
 
-    ~libusb_managed_send_buffer_impl(void){
-        if (!_committed) {
-            _lut->length = 0;
-            _lut->actual_length = 0;
-            _endpoint->submit(_lut);
-        }
+private:
+    void release(libusb_transfer *lut){
+        _recv_ep->submit(lut);
     }
 
-    ssize_t commit(size_t num_bytes)
-    {
-        if (_committed) {
-            std::cerr << "UHD: send buffer already committed" << std::endl;
-            return 0;
-        }
-
-        UHD_ASSERT_THROW(num_bytes <= boost::asio::buffer_size(_buff));
-
-        _lut->length = num_bytes;
-        _lut->actual_length = 0;
-
+    void commit(libusb_transfer *lut, size_t num_bytes){
+        lut->length = num_bytes;
         try{
-            _endpoint->submit(_lut);
-            _committed = true;
-            return num_bytes;
+            _send_ep->submit(lut);
         }
         catch(const std::exception &e){
             std::cerr << "Error in commit: " << e.what() << std::endl;
-            return -1;
         }
     }
 
-private:
-    const boost::asio::mutable_buffer &get(void) const{
-        return _buff;
-    }
-
-    libusb_transfer *_lut;
-    usb_endpoint::sptr _endpoint;
-    const boost::asio::mutable_buffer _buff;
-    bool _committed;
-};
-
-
-/***********************************************************************
- * USB zero_copy device class
- **********************************************************************/
-class libusb_zero_copy_impl : public usb_zero_copy
-{
-private:
     libusb::device_handle::sptr _handle;
+    size_t _recv_xfer_size, _send_xfer_size;
     size_t _recv_num_frames, _send_num_frames;
     usb_endpoint::sptr _recv_ep, _send_ep;
-
-public:
-    typedef boost::shared_ptr<libusb_zero_copy_impl> sptr;
-
-    libusb_zero_copy_impl(
-        libusb::device_handle::sptr handle,
-        unsigned int recv_endpoint, unsigned int send_endpoint,
-        size_t recv_xfer_size, size_t recv_num_xfers,
-        size_t send_xfer_size, size_t send_num_xfers
-    );
-
-    managed_recv_buffer::sptr get_recv_buff(double);
-    managed_send_buffer::sptr get_send_buff(double);
-
-    size_t get_num_recv_frames(void) const { return _recv_num_frames; }
-    size_t get_num_send_frames(void) const { return _send_num_frames; }
 };
 
 /*
@@ -426,7 +339,9 @@ libusb_zero_copy_impl::libusb_zero_copy_impl(
     UHD_ASSERT_THROW(send_xfer_size % 512 == 0);
 
     //store the num xfers for the num frames count
+    _recv_xfer_size = recv_xfer_size;
     _recv_num_frames = recv_num_xfers;
+    _send_xfer_size = send_xfer_size;
     _send_num_frames = send_num_xfers;
 
     _handle->claim_interface(2 /*in interface*/);
@@ -461,9 +376,10 @@ managed_recv_buffer::sptr libusb_zero_copy_impl::get_recv_buff(double timeout){
         return managed_recv_buffer::sptr();
     }
     else {
-        return managed_recv_buffer::sptr(
-            new libusb_managed_recv_buffer_impl(lut,
-                                                _recv_ep));
+        return managed_recv_buffer::make_safe(
+            boost::asio::const_buffer(lut->buffer, lut->actual_length),
+            boost::bind(&libusb_zero_copy_impl::release, this, lut)
+        );
     }
 }
 
@@ -480,9 +396,10 @@ managed_send_buffer::sptr libusb_zero_copy_impl::get_send_buff(double timeout){
         return managed_send_buffer::sptr();
     }
     else {
-        return managed_send_buffer::sptr(
-            new libusb_managed_send_buffer_impl(lut,
-                                                _send_ep));
+        return managed_send_buffer::make_safe(
+            boost::asio::mutable_buffer(lut->buffer, _send_xfer_size),
+            boost::bind(&libusb_zero_copy_impl::commit, this, lut, _1)
+        );
     }
 }
 
diff --git a/host/lib/transport/vrt_packet_handler.hpp b/host/lib/transport/vrt_packet_handler.hpp
index e11afff30..939517411 100644
--- a/host/lib/transport/vrt_packet_handler.hpp
+++ b/host/lib/transport/vrt_packet_handler.hpp
@@ -343,9 +343,7 @@ template <typename T> UHD_INLINE T get_context_code(
 
             //commit the samples to the zero-copy interface
             size_t num_bytes_total = (vrt_header_offset_words32+if_packet_info.num_packet_words32)*sizeof(boost::uint32_t);
-            if (send_buffs[i]->commit(num_bytes_total) < ssize_t(num_bytes_total)){
-                std::cerr << "commit to send buffer returned less than commit size" << std::endl;
-            }
+            send_buffs[i]->commit(num_bytes_total);
         }
         return num_samps;
     }
diff --git a/host/lib/transport/zero_copy.cpp b/host/lib/transport/zero_copy.cpp
index dfb65951f..a5a864a04 100644
--- a/host/lib/transport/zero_copy.cpp
+++ b/host/lib/transport/zero_copy.cpp
@@ -16,32 +16,35 @@
 //
 
 #include <uhd/transport/zero_copy.hpp>
-#include <boost/cstdint.hpp>
-#include <boost/function.hpp>
-#include <boost/bind.hpp>
 
 using namespace uhd::transport;
 
 /***********************************************************************
- * The pure-virtual deconstructor needs an implementation to be happy
+ * Safe managed receive buffer
  **********************************************************************/
-managed_recv_buffer::~managed_recv_buffer(void){
+static void release_nop(void){
     /* NOP */
 }
 
-/***********************************************************************
- * Phony zero-copy recv interface implementation
- **********************************************************************/
-
-//! phony zero-copy recv buffer implementation
-class managed_recv_buffer_impl : public managed_recv_buffer{
+class safe_managed_receive_buffer : public managed_recv_buffer{
 public:
-    managed_recv_buffer_impl(const boost::asio::const_buffer &buff) : _buff(buff){
+    safe_managed_receive_buffer(
+        const boost::asio::const_buffer &buff,
+        const release_fcn_t &release_fcn
+    ):
+        _buff(buff), _release_fcn(release_fcn)
+    {
         /* NOP */
     }
 
-    ~managed_recv_buffer_impl(void){
-        delete [] this->cast<const boost::uint8_t *>();
+    ~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:
@@ -50,64 +53,42 @@ private:
     }
 
     const boost::asio::const_buffer _buff;
+    release_fcn_t _release_fcn;
 };
 
-//! phony zero-copy recv interface implementation
-struct phony_zero_copy_recv_if::impl{
-    impl(size_t max_buff_size) : max_buff_size(max_buff_size){
-        /* NOP */
-    }
-    size_t max_buff_size;
-};
-
-phony_zero_copy_recv_if::phony_zero_copy_recv_if(size_t max_buff_size){
-    _impl = UHD_PIMPL_MAKE(impl, (max_buff_size));
-}
-
-phony_zero_copy_recv_if::~phony_zero_copy_recv_if(void){
-    /* NOP */
-}
-
-managed_recv_buffer::sptr phony_zero_copy_recv_if::get_recv_buff(double timeout){
-    //allocate memory
-    boost::uint8_t *recv_mem = new boost::uint8_t[_impl->max_buff_size];
-
-    //call recv() with timeout option
-    ssize_t num_bytes = this->recv(boost::asio::buffer(recv_mem, _impl->max_buff_size), timeout);
-
-    if (num_bytes <= 0) return managed_recv_buffer::sptr(); //NULL sptr
-
-    //create a new managed buffer to house the data
-    return managed_recv_buffer::sptr(
-        new managed_recv_buffer_impl(boost::asio::buffer(recv_mem, num_bytes))
-    );
+managed_recv_buffer::sptr managed_recv_buffer::make_safe(
+    const boost::asio::const_buffer &buff,
+    const release_fcn_t &release_fcn
+){
+    return sptr(new safe_managed_receive_buffer(buff, release_fcn));
 }
 
 /***********************************************************************
- * Phony zero-copy send interface implementation
+ * Safe managed send buffer
  **********************************************************************/
+static void commit_nop(size_t){
+    /* NOP */
+}
 
-//! phony zero-copy send buffer implementation
-class managed_send_buffer_impl : public managed_send_buffer{
+class safe_managed_send_buffer : public managed_send_buffer{
 public:
-    typedef boost::function<ssize_t(const boost::asio::const_buffer &)> send_fcn_t;
-
-    managed_send_buffer_impl(
+    safe_managed_send_buffer(
         const boost::asio::mutable_buffer &buff,
-        const send_fcn_t &send_fcn
+        const commit_fcn_t &commit_fcn
     ):
-        _buff(buff),
-        _send_fcn(send_fcn)
+        _buff(buff), _commit_fcn(commit_fcn)
     {
         /* NOP */
     }
 
-    ~managed_send_buffer_impl(void){
-        /* NOP */
+    ~safe_managed_send_buffer(void){
+        _commit_fcn(0);
     }
 
-    ssize_t commit(size_t num_bytes){
-        return _send_fcn(boost::asio::buffer(_buff, num_bytes));
+    void commit(size_t num_bytes){
+        commit_fcn_t commit_fcn = _commit_fcn;
+        _commit_fcn = &commit_nop;
+        return commit_fcn(num_bytes);
     }
 
 private:
@@ -116,28 +97,12 @@ private:
     }
 
     const boost::asio::mutable_buffer _buff;
-    const send_fcn_t                  _send_fcn;
-};
-
-//! phony zero-copy send interface implementation
-struct phony_zero_copy_send_if::impl{
-    boost::uint8_t *send_mem;
-    managed_send_buffer::sptr send_buff;
+    commit_fcn_t _commit_fcn;
 };
 
-phony_zero_copy_send_if::phony_zero_copy_send_if(size_t max_buff_size){
-    _impl = UHD_PIMPL_MAKE(impl, ());
-    _impl->send_mem = new boost::uint8_t[max_buff_size];
-    _impl->send_buff = managed_send_buffer::sptr(new managed_send_buffer_impl(
-        boost::asio::buffer(_impl->send_mem, max_buff_size),
-        boost::bind(&phony_zero_copy_send_if::send, this, _1)
-    ));
-}
-
-phony_zero_copy_send_if::~phony_zero_copy_send_if(void){
-    delete [] _impl->send_mem;
-}
-
-managed_send_buffer::sptr phony_zero_copy_send_if::get_send_buff(double){
-    return _impl->send_buff; //FIXME there is only ever one send buff, we assume that the caller doesnt hang onto these
+safe_managed_send_buffer::sptr managed_send_buffer::make_safe(
+    const boost::asio::mutable_buffer &buff,
+    const commit_fcn_t &commit_fcn
+){
+    return sptr(new safe_managed_send_buffer(buff, commit_fcn));
 }
diff --git a/host/lib/usrp/usrp1/io_impl.cpp b/host/lib/usrp/usrp1/io_impl.cpp
index 8d9c68961..676b1536a 100644
--- a/host/lib/usrp/usrp1/io_impl.cpp
+++ b/host/lib/usrp/usrp1/io_impl.cpp
@@ -34,35 +34,6 @@ using namespace uhd::transport;
 namespace asio = boost::asio;
 
 /***********************************************************************
- * Pseudo send buffer implementation
- **********************************************************************/
-class pseudo_managed_send_buffer : public managed_send_buffer{
-public:
-
-    pseudo_managed_send_buffer(
-        const boost::asio::mutable_buffer &buff,
-        const boost::function<ssize_t(size_t)> &commit
-    ):
-        _buff(buff),
-        _commit(commit)
-    {
-        /* NOP */
-    }
-
-    ssize_t commit(size_t num_bytes){
-        return _commit(num_bytes);
-    }
-
-private:
-    const boost::asio::mutable_buffer &get(void) const{
-        return _buff;
-    }
-
-    const boost::asio::mutable_buffer      _buff;
-    const boost::function<ssize_t(size_t)> _commit;
-};
-
-/***********************************************************************
  * IO Implementation Details
  **********************************************************************/
 struct usrp1_impl::io_impl{
@@ -96,9 +67,9 @@ struct usrp1_impl::io_impl{
     size_t num_bytes_committed;
     double send_timeout;
     boost::uint8_t pseudo_buff[BYTES_PER_PACKET];
-    ssize_t phony_commit_pseudo_buff(size_t num_bytes);
-    ssize_t phony_commit_send_buff(size_t num_bytes);
-    ssize_t commit_send_buff(void);
+    void phony_commit_pseudo_buff(size_t num_bytes);
+    void phony_commit_send_buff(size_t num_bytes);
+    void commit_send_buff(void);
     void flush_send_buff(void);
     bool get_send_buffs(vrt_packet_handler::managed_send_buffs_t &, double);
 
@@ -119,28 +90,25 @@ struct usrp1_impl::io_impl{
  *   The first loop iteration will fill the remainder of the send buffer.
  *   The second loop iteration will empty the pseudo buffer remainder.
  */
-ssize_t usrp1_impl::io_impl::phony_commit_pseudo_buff(size_t num_bytes){
+void usrp1_impl::io_impl::phony_commit_pseudo_buff(size_t num_bytes){
     size_t bytes_to_copy = num_bytes, bytes_copied = 0;
     while(bytes_to_copy){
         size_t bytes_copied_here = std::min(bytes_to_copy, get_send_mem_size());
         std::memcpy(get_send_mem_ptr(), pseudo_buff + bytes_copied, bytes_copied_here);
-        ssize_t ret = phony_commit_send_buff(bytes_copied_here);
-        if (ret < 0) return ret;
-        bytes_to_copy -= ret;
-        bytes_copied += ret;
+        phony_commit_send_buff(bytes_copied_here);
+        bytes_to_copy -= bytes_copied_here;
+        bytes_copied += bytes_copied_here;
     }
-    return bytes_copied;
 }
 
 /*!
  * Accept a commit of num bytes to the send buffer.
  * Conditionally commit the send buffer if full.
  */
-ssize_t usrp1_impl::io_impl::phony_commit_send_buff(size_t num_bytes){
+void usrp1_impl::io_impl::phony_commit_send_buff(size_t num_bytes){
     num_bytes_committed += num_bytes;
-    if (num_bytes_committed != send_buff->size()) return num_bytes;
-    ssize_t ret = commit_send_buff();
-    return (ret < 0)? ret : num_bytes;
+    if (num_bytes_committed != send_buff->size()) return;
+    commit_send_buff();
 }
 
 /*!
@@ -158,11 +126,10 @@ void usrp1_impl::io_impl::flush_send_buff(void){
  * Perform an actual commit on the send buffer:
  * Commit the contents of the send buffer and request a new buffer.
  */
-ssize_t usrp1_impl::io_impl::commit_send_buff(void){
-    ssize_t ret = send_buff->commit(num_bytes_committed);
+void usrp1_impl::io_impl::commit_send_buff(void){
+    send_buff->commit(num_bytes_committed);
     send_buff = data_transport->get_send_buff(send_timeout);
     num_bytes_committed = 0;
-    return ret;
 }
 
 bool usrp1_impl::io_impl::get_send_buffs(
@@ -173,17 +140,17 @@ bool usrp1_impl::io_impl::get_send_buffs(
 
     //not enough bytes free -> use the pseudo buffer
     if (get_send_mem_size() < BYTES_PER_PACKET){
-        buffs[0] = managed_send_buffer::sptr(new pseudo_managed_send_buffer(
+        buffs[0] = managed_send_buffer::make_safe(
             boost::asio::buffer(pseudo_buff),
             boost::bind(&usrp1_impl::io_impl::phony_commit_pseudo_buff, this, _1)
-        ));
+        );
     }
     //otherwise use the send buffer offset by the bytes written
     else{
-        buffs[0] = managed_send_buffer::sptr(new pseudo_managed_send_buffer(
+        buffs[0] = managed_send_buffer::make_safe(
             boost::asio::buffer(get_send_mem_ptr(), get_send_mem_size()),
             boost::bind(&usrp1_impl::io_impl::phony_commit_send_buff, this, _1)
-        ));
+        );
     }
 
     return buffs[0].get() != NULL;