udp: simplfy zero copy asio overhead with less shared_from_this, and timed waits when not needed

3 files changed, 43 insertions, 21 deletions

diff --git a/host/include/uhd/transport/bounded_buffer.hpp b/host/include/uhd/transport/bounded_buffer.hpp
index aca93b071..33ded8cfd 100644
--- a/host/include/uhd/transport/bounded_buffer.hpp
+++ b/host/include/uhd/transport/bounded_buffer.hpp
@@ -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
@@ -66,6 +66,13 @@ namespace uhd{ namespace transport{
         virtual bool push_with_timed_wait(const elem_type &elem, double timeout) = 0;
 
         /*!
+         * Pop an element from the bounded_buffer immediately.
+         * \param elem the element reference pop to
+         * \return false when the bounded_buffer is empty
+         */
+        virtual bool pop_with_haste(elem_type &elem) = 0;
+
+        /*!
          * Pop an element from the bounded_buffer.
          * Wait until the bounded_buffer becomes non-empty.
          * \param elem the element reference pop to
diff --git a/host/include/uhd/transport/bounded_buffer.ipp b/host/include/uhd/transport/bounded_buffer.ipp
index 4fbe3f085..7bf182ee3 100644
--- a/host/include/uhd/transport/bounded_buffer.ipp
+++ b/host/include/uhd/transport/bounded_buffer.ipp
@@ -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
@@ -72,6 +72,15 @@ namespace uhd{ namespace transport{ namespace{ /*anon*/
             return true;
         }
 
+        UHD_INLINE bool pop_with_haste(elem_type &elem){
+            boost::mutex::scoped_lock lock(_mutex);
+            if(_buffer.empty()) return false;
+            elem = this->pop_back();
+            lock.unlock();
+            _full_cond.notify_one();
+            return true;
+        }
+
         UHD_INLINE void pop_with_wait(elem_type &elem){
             boost::mutex::scoped_lock lock(_mutex);
             _empty_cond.wait(lock, _not_empty_fcn);
diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp
index 697e172cd..f083b97d8 100644
--- a/host/lib/transport/udp_zero_copy_asio.cpp
+++ b/host/lib/transport/udp_zero_copy_asio.cpp
@@ -34,7 +34,7 @@ namespace asio = boost::asio;
 
 //Define this to the the boost async io calls to perform receive.
 //Otherwise, get_recv_buff uses a blocking receive with timeout.
-#define USE_ASIO_ASYNC_RECV
+//#define USE_ASIO_ASYNC_RECV
 
 //Define this to the the boost async io calls to perform send.
 //Otherwise, the commit callback uses a blocking send.
@@ -138,8 +138,7 @@ public:
 
     //! handle a recv callback -> push the filled memory into the fifo
     UHD_INLINE void handle_recv(void *mem, size_t len){
-        boost::this_thread::disable_interruption di; //disable because the wait can throw
-        _pending_recv_buffs->push_with_wait(boost::asio::buffer(mem, len));
+        _pending_recv_buffs->push_with_pop_on_full(boost::asio::buffer(mem, len));
     }
 
     ////////////////////////////////////////////////////////////////////
@@ -152,8 +151,7 @@ public:
         if (_pending_recv_buffs->pop_with_timed_wait(buff, timeout)){
             return managed_recv_buffer::make_safe(
                 buff, boost::bind(
-                    &udp_zero_copy_asio_impl::release,
-                    shared_from_this(),
+                    &udp_zero_copy_asio_impl::release, this,
                     asio::buffer_cast<void*>(buff)
                 )
             );
@@ -177,7 +175,6 @@ public:
     #else /*USE_ASIO_ASYNC_RECV*/
     ////////////////////////////////////////////////////////////////////
     managed_recv_buffer::sptr get_recv_buff(double timeout){
-        boost::this_thread::disable_interruption di; //disable because the wait can throw
         asio::mutable_buffer buff;
 
         //setup timeval for timeout
@@ -190,11 +187,11 @@ public:
         FD_ZERO(&rset);
         FD_SET(_sock_fd, &rset);
 
-        //call select to perform timed wait and grab an available buffer with wait
+        //call select to perform timed wait and grab an available buffer now
         //if the condition is true, call receive and return the managed buffer
         if (
-            ::select(_sock_fd+1, &rset, NULL, NULL, &tv) > 0 and
-            _pending_recv_buffs->pop_with_timed_wait(buff, timeout)
+            ::select(_sock_fd+1, &rset, NULL, NULL, &tv) > 0
+            and _pending_recv_buffs->pop_with_haste(buff)
         ){
             return managed_recv_buffer::make_safe(
                 asio::buffer(
@@ -202,8 +199,7 @@ public:
                     _socket->receive(asio::buffer(buff))
                 ),
                 boost::bind(
-                    &udp_zero_copy_asio_impl::release,
-                    shared_from_this(),
+                    &udp_zero_copy_asio_impl::release, this,
                     asio::buffer_cast<void*>(buff)
                 )
             );
@@ -212,7 +208,6 @@ public:
     }
 
     void release(void *mem){
-        boost::this_thread::disable_interruption di; //disable because the wait can throw
         handle_recv(mem, this->get_recv_frame_size());
     }
 
@@ -225,10 +220,12 @@ public:
 
     //! handle a send callback -> push the emptied memory into the fifo
     UHD_INLINE void handle_send(void *mem){
-        boost::this_thread::disable_interruption di; //disable because the wait can throw
-        _pending_send_buffs->push_with_wait(boost::asio::buffer(mem, this->get_send_frame_size()));
+        _pending_send_buffs->push_with_pop_on_full(boost::asio::buffer(mem, this->get_send_frame_size()));
     }
 
+    ////////////////////////////////////////////////////////////////////
+    #ifdef USE_ASIO_ASYNC_SEND
+    ////////////////////////////////////////////////////////////////////
     //! pop an empty send buffer off of the fifo and bind with the commit callback
     managed_send_buffer::sptr get_send_buff(double timeout){
         boost::this_thread::disable_interruption di; //disable because the wait can throw
@@ -236,8 +233,7 @@ public:
         if (_pending_send_buffs->pop_with_timed_wait(buff, timeout)){
             return managed_send_buffer::make_safe(
                 buff, boost::bind(
-                    &udp_zero_copy_asio_impl::commit,
-                    shared_from_this(),
+                    &udp_zero_copy_asio_impl::commit, this,
                     asio::buffer_cast<void*>(buff), _1
                 )
             );
@@ -245,9 +241,6 @@ public:
         return managed_send_buffer::sptr();
     }
 
-    ////////////////////////////////////////////////////////////////////
-    #ifdef USE_ASIO_ASYNC_SEND
-    ////////////////////////////////////////////////////////////////////
     //! commit a send buffer -> start an async send on the buffer
     void commit(void *mem, size_t len){
         _socket->async_send(
@@ -262,6 +255,19 @@ public:
     ////////////////////////////////////////////////////////////////////
     #else /*USE_ASIO_ASYNC_SEND*/
     ////////////////////////////////////////////////////////////////////
+    managed_send_buffer::sptr get_send_buff(double){
+        asio::mutable_buffer buff;
+        if (_pending_send_buffs->pop_with_haste(buff)){
+            return managed_send_buffer::make_safe(
+                buff, boost::bind(
+                    &udp_zero_copy_asio_impl::commit, this,
+                    asio::buffer_cast<void*>(buff), _1
+                )
+            );
+        }
+        return managed_send_buffer::sptr();
+    }
+
     void commit(void *mem, size_t len){
         _socket->send(asio::buffer(mem, len));
         handle_send(mem);