1 files changed, 45 insertions, 2 deletions

diff --git a/host/lib/transport/usb_zero_copy_wrapper.cpp b/host/lib/transport/usb_zero_copy_wrapper.cpp
index 87e001fed..843b20f11 100644
--- a/host/lib/transport/usb_zero_copy_wrapper.cpp
+++ b/host/lib/transport/usb_zero_copy_wrapper.cpp
@@ -20,13 +20,19 @@
 #include <uhd/transport/buffer_pool.hpp>
 #include <uhd/utils/byteswap.hpp>
 #include <uhd/utils/msg.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <boost/foreach.hpp>
 #include <boost/make_shared.hpp>
+#include <boost/thread/mutex.hpp>
+#include <boost/thread/condition_variable.hpp>
+#include <boost/bind.hpp>
 #include <vector>
 #include <iostream>
 
 using namespace uhd::transport;
 
+static const boost::posix_time::time_duration AUTOFLUSH_TIMEOUT(boost::posix_time::milliseconds(1));
+
 /***********************************************************************
  * USB zero copy wrapper - managed receive buffer
  **********************************************************************/
@@ -53,11 +59,13 @@ public:
         //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)
+        const size_t words32 = (uhd::wtohx(mem32[0]) & 0xffff); //length in words32 (from VRT header)
+        const size_t len = words32*sizeof(boost::uint32_t); //length in bytes
 
         //check if this receive buffer has been exhausted
         offset_bytes += len;
         if (offset_bytes >= mrb->size()) mrb.reset(); //drop caller's ref
+        else if (uhd::wtohx(mem32[words32]) == 0) mrb.reset();
 
         return make(this, mem, len);
     }
@@ -73,9 +81,16 @@ private:
 class usb_zero_copy_wrapper_msb : public managed_send_buffer{
 public:
     usb_zero_copy_wrapper_msb(const usb_zero_copy::sptr internal, const size_t fragmentation_size):
-        _internal(internal), _fragmentation_size(fragmentation_size){/*NOP*/}
+        _internal(internal), _fragmentation_size(fragmentation_size)
+    {
+        _ok_to_auto_flush = false;
+        _task = uhd::task::make(boost::bind(&usb_zero_copy_wrapper_msb::auto_flush, this));
+    }
 
     void release(void){
+        boost::mutex::scoped_lock lock(_mutex);
+        _ok_to_auto_flush = true;
+
         //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];
 
@@ -88,16 +103,23 @@ public:
         if (eop or full){
             _last_send_buff->commit(_bytes_in_buffer);
             _last_send_buff.reset();
+
+            //notify the auto-flusher to restart its timed_wait
+            lock.unlock(); _cond.notify_one();
         }
     }
 
     UHD_INLINE sptr get_new(const double timeout){
+        boost::mutex::scoped_lock lock(_mutex);
+        _ok_to_auto_flush = false;
+
         if (not _last_send_buff){
             _last_send_buff = _internal->get_send_buff(timeout);
             if (not _last_send_buff) return sptr();
             _mem_buffer_tip = _last_send_buff->cast<char *>();
             _bytes_in_buffer = 0;
         }
+
         return make(this, _mem_buffer_tip, _fragmentation_size);
     }
 
@@ -107,6 +129,27 @@ private:
     managed_send_buffer::sptr _last_send_buff;
     size_t _bytes_in_buffer;
     char *_mem_buffer_tip;
+
+    //private variables for auto flusher
+    boost::mutex _mutex;
+    boost::condition_variable _cond;
+    uhd::task::sptr _task;
+    bool _ok_to_auto_flush;
+
+    /*!
+     * The auto flusher ensures that buffers are force committed when
+     * the user has not called get_new() within a certain time window.
+     */
+    void auto_flush(void)
+    {
+        boost::mutex::scoped_lock lock(_mutex);
+        const bool timeout = not _cond.timed_wait(lock, AUTOFLUSH_TIMEOUT);
+        if (timeout and _ok_to_auto_flush and _last_send_buff and _bytes_in_buffer != 0)
+        {
+            _last_send_buff->commit(_bytes_in_buffer);
+            _last_send_buff.reset();
+        }
+    }
 };
 
 /***********************************************************************