uhd: switch the reusable barrier to condition variables

This allows the converter threads in a multi-threaded streamer to wait quietly.
In addition, the use of two barriers in the packet handlers was reduced to one,
by adding a simple exit barrier inside the reusable barrier's wait method.

2 files changed, 12 insertions, 16 deletions

diff --git a/host/lib/transport/super_recv_packet_handler.hpp b/host/lib/transport/super_recv_packet_handler.hpp
index 7a1972690..5a75d5f0d 100644
--- a/host/lib/transport/super_recv_packet_handler.hpp
+++ b/host/lib/transport/super_recv_packet_handler.hpp
@@ -1,5 +1,5 @@
 //
-// Copyright 2011-2012 Ettus Research LLC
+// Copyright 2011-2013 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
@@ -79,8 +79,7 @@ public:
     }
 
     ~recv_packet_handler(void){
-        _task_barrier_entry.interrupt();
-        _task_barrier_exit.interrupt();
+        _task_barrier.interrupt();
         _task_handlers.clear();
     }
 
@@ -91,8 +90,7 @@ public:
         _props.resize(size);
         //re-initialize all buffers infos by re-creating the vector
         _buffers_infos = std::vector<buffers_info_type>(4, buffers_info_type(size));
-        _task_barrier_entry.resize(size);
-        _task_barrier_exit.resize(size);
+        _task_barrier.resize(size);
         _task_handlers.resize(size);
         for (size_t i = 1/*skip 0*/; i < size; i++){
             _task_handlers[i] = task::make(boost::bind(&recv_packet_handler::converter_thread_task, this, i));
@@ -561,7 +559,7 @@ private:
      ******************************************************************/
     UHD_INLINE void converter_thread_task(const size_t index)
     {
-        _task_barrier_entry.wait();
+        _task_barrier.wait();
 
         //shortcut references to local data structures
         buffers_info_type &buff_info = get_curr_buffer_info();
@@ -587,11 +585,11 @@ private:
             info.buff.reset(); //effectively a release
         }
 
-        _task_barrier_exit.wait();
+        if (index == 0) _task_barrier.wait_others();
     }
 
     //! Shared variables for the worker threads
-    reusable_barrier _task_barrier_entry, _task_barrier_exit;
+    reusable_barrier _task_barrier;
     std::vector<task::sptr> _task_handlers;
     size_t _convert_nsamps;
     const rx_streamer::buffs_type *_convert_buffs;
diff --git a/host/lib/transport/super_send_packet_handler.hpp b/host/lib/transport/super_send_packet_handler.hpp
index 74e893e67..726742327 100644
--- a/host/lib/transport/super_send_packet_handler.hpp
+++ b/host/lib/transport/super_send_packet_handler.hpp
@@ -1,5 +1,5 @@
 //
-// Copyright 2011-2012 Ettus Research LLC
+// Copyright 2011-2013 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
@@ -61,8 +61,7 @@ public:
     }
 
     ~send_packet_handler(void){
-        _task_barrier_entry.interrupt();
-        _task_barrier_exit.interrupt();
+        _task_barrier.interrupt();
         _task_handlers.clear();
     }
 
@@ -73,8 +72,7 @@ public:
         _props.resize(size);
         static const boost::uint64_t zero = 0;
         _zero_buffs.resize(size, &zero);
-        _task_barrier_entry.resize(size);
-        _task_barrier_exit.resize(size);
+        _task_barrier.resize(size);
         _task_handlers.resize(size);
         for (size_t i = 1/*skip 0*/; i < size; i++){
             _task_handlers[i] = task::make(boost::bind(&send_packet_handler::converter_thread_task, this, i));
@@ -272,7 +270,7 @@ private:
      ******************************************************************/
     UHD_INLINE void converter_thread_task(const size_t index)
     {
-        _task_barrier_entry.wait();
+        _task_barrier.wait();
 
         //shortcut references to local data structures
         managed_send_buffer::sptr &buff = _props[index].buff;
@@ -302,11 +300,11 @@ private:
         buff->commit(num_vita_words32*sizeof(boost::uint32_t));
         buff.reset(); //effectively a release
 
-        _task_barrier_exit.wait();
+        if (index == 0) _task_barrier.wait_others();
     }
 
     //! Shared variables for the worker threads
-    reusable_barrier _task_barrier_entry, _task_barrier_exit;
+    reusable_barrier _task_barrier;
     std::vector<task::sptr> _task_handlers;
     size_t _convert_nsamps;
     const tx_streamer::buffs_type *_convert_buffs;