2 files changed, 111 insertions, 10 deletions

diff --git a/host/lib/utils/paths.cpp b/host/lib/utils/paths.cpp
index f9d8b613c..3e2bea1c6 100644
--- a/host/lib/utils/paths.cpp
+++ b/host/lib/utils/paths.cpp
@@ -34,8 +34,19 @@
 namespace fs = boost::filesystem;
 
 /***********************************************************************
+ * Get a list of paths for an environment variable
+ **********************************************************************/
+static std::string get_env_var(const std::string &var_name, const std::string &def_val = ""){
+    const char *var_value_ptr = std::getenv(var_name.c_str());
+    return (var_value_ptr == NULL)? def_val : var_value_ptr;
+}
+
+static std::vector<fs::path> get_env_paths(const std::string &var_name){
+
+/***********************************************************************
  * Determine the paths separator
  **********************************************************************/
+
 #ifdef UHD_PLATFORM_WIN32
     static const std::string env_path_sep = ";";
 #else
@@ -46,16 +57,6 @@ namespace fs = boost::filesystem;
     boost::tokenizer<boost::char_separator<char> > \
     (inp, boost::char_separator<char>(env_path_sep.c_str()))
 
-/***********************************************************************
- * Get a list of paths for an environment variable
- **********************************************************************/
-static std::string get_env_var(const std::string &var_name, const std::string &def_val = ""){
-    const char *var_value_ptr = std::getenv(var_name.c_str());
-    return (var_value_ptr == NULL)? def_val : var_value_ptr;
-}
-
-static std::vector<fs::path> get_env_paths(const std::string &var_name){
-
     std::string var_value = get_env_var(var_name);
 
     //convert to filesystem path, filter blank paths
@@ -85,6 +86,7 @@ std::vector<fs::path> get_image_paths(void){
 std::vector<fs::path> get_module_paths(void){
     std::vector<fs::path> paths = get_env_paths("UHD_MODULE_PATH");
     paths.push_back(fs::path(uhd::get_pkg_path()) / UHD_LIB_DIR / "uhd" / "modules");
+    paths.push_back(fs::path(uhd::get_pkg_path()) / "share" / "uhd" / "modules");
     return paths;
 }
 
diff --git a/host/lib/utils/tasks.cpp b/host/lib/utils/tasks.cpp
index 1f735de06..08c32a5fb 100644
--- a/host/lib/utils/tasks.cpp
+++ b/host/lib/utils/tasks.cpp
@@ -16,11 +16,13 @@
 //
 
 #include <uhd/utils/tasks.hpp>
+#include <uhd/utils/msg_task.hpp>
 #include <uhd/utils/msg.hpp>
 #include <boost/thread/thread.hpp>
 #include <boost/thread/barrier.hpp>
 #include <exception>
 #include <iostream>
+#include <vector>
 
 using namespace uhd;
 
@@ -80,3 +82,100 @@ private:
 task::sptr task::make(const task_fcn_type &task_fcn){
     return task::sptr(new task_impl(task_fcn));
 }
+
+/*
+ * During shutdown pointers to queues for radio_ctrl_core might not be available anymore.
+ * msg_task_impl provides a dump_queue for such messages.
+ * ctrl_cores can check this queue for stranded messages.
+ */
+
+class msg_task_impl : public msg_task{
+public:
+
+    msg_task_impl(const task_fcn_type &task_fcn):
+        _spawn_barrier(2)
+    {
+        _thread_group.create_thread(boost::bind(&msg_task_impl::task_loop, this, task_fcn));
+        _spawn_barrier.wait();
+    }
+
+    ~msg_task_impl(void){
+        _running = false;
+        _thread_group.interrupt_all();
+        _thread_group.join_all();
+    }
+
+    /*
+     * Returns the first message for the given SID.
+     * This way a radio_ctrl_core doesn't have to die in timeout but can check for stranded messages here.
+     * This might happen during shutdown when dtors are called.
+     * See also: comments in b200_io_impl->handle_async_task
+     */
+    msg_payload_t get_msg_from_dump_queue(boost::uint32_t sid)
+    {
+        boost::mutex::scoped_lock lock(_mutex);
+        msg_payload_t b;
+        for (size_t i = 0; i < _dump_queue.size(); i++) {
+            if (sid == _dump_queue[i].first) {
+                b = _dump_queue[i].second;
+                _dump_queue.erase(_dump_queue.begin() + i);
+                break;
+            }
+        }
+        return b;
+    }
+
+private:
+
+    void task_loop(const task_fcn_type &task_fcn){
+        _running = true;
+        _spawn_barrier.wait();
+
+        try{
+            while (_running){
+            	boost::optional<msg_type_t> buff = task_fcn();
+            	if(buff != boost::none){
+            	    /*
+            	     * If a message gets stranded it is returned by task_fcn and then pushed to the dump_queue.
+            	     * This way ctrl_cores can check dump_queue for missing messages.
+            	     */
+            	    boost::mutex::scoped_lock lock(_mutex);
+            	    _dump_queue.push_back(buff.get() );
+            	}
+            }
+        }
+        catch(const boost::thread_interrupted &){
+            //this is an ok way to exit the task loop
+        }
+        catch(const std::exception &e){
+            do_error_msg(e.what());
+        }
+        catch(...){
+            //FIXME
+            //Unfortunately, this is also an ok way to end a task,
+            //because on some systems boost throws uncatchables.
+        }
+    }
+
+    void do_error_msg(const std::string &msg){
+        UHD_MSG(error)
+            << "An unexpected exception was caught in a task loop." << std::endl
+            << "The task loop will now exit, things may not work." << std::endl
+            << msg << std::endl
+        ;
+    }
+
+    boost::mutex _mutex;
+    boost::thread_group _thread_group;
+    boost::barrier _spawn_barrier;
+    bool _running;
+
+    /*
+     * This queue holds stranded messages until a radio_ctrl_core grabs them via 'get_msg_from_dump_queue'.
+     */
+    std::vector <msg_type_t> _dump_queue;
+};
+
+msg_task::sptr msg_task::make(const task_fcn_type &task_fcn){
+    return msg_task::sptr(new msg_task_impl(task_fcn));
+}