From 58cb4bb733fc14de0ca520620cfcb689e2155797 Mon Sep 17 00:00:00 2001
From: Johannes Meyer <johannes@intermodalics.eu>
Date: Mon, 13 Apr 2015 16:39:35 +0200
Subject: [PATCH] engine/activity: fixed blocking calls to
 ExecutionEngine::process()

Broadcasting condition variables without holding the associated mutex can block for some reasons.
This patch adds MutexLocks whereever the msg_cond is triggered.

From http://pubs.opengroup.org/onlinepubs/009695399/functions/pthread_cond_broadcast.html:
> The pthread_cond_broadcast() or pthread_cond_signal() functions may be called by a thread whether
> or not it currently owns the mutex that threads calling pthread_cond_wait() or pthread_cond_timedwait()
> have associated with the condition variable during their waits; however, if predictable scheduling
> behavior is required, then that mutex shall be locked by the thread calling pthread_cond_broadcast() or
> pthread_cond_signal().

This patch also clears the Thread active flag in Thread::terminate() is called and forces the thread
function to be terminated in the destructor of class Activity. This solves some issues with pure virtual
method calls during the destruction phase if threads are still running and triggering each other.

Signed-off-by: Johannes Meyer <johannes@intermodalics.eu>
---
 rtt/Activity.cpp        | 10 ++++++++--
 rtt/ExecutionEngine.cpp | 24 ++++++++++++++----------
 rtt/os/Thread.cpp       |  1 +
 3 files changed, 23 insertions(+), 12 deletions(-)

diff --git a/rtt/Activity.cpp b/rtt/Activity.cpp
index 230f22ac7..2bcaf995d 100644
--- a/rtt/Activity.cpp
+++ b/rtt/Activity.cpp
@@ -103,6 +103,10 @@ namespace RTT
     Activity::~Activity()
     {
         stop();
+
+        // We need to join the activity's thread before destruction as the thread function might still
+        // access member variables. Activity::stop() does not guarantuee to stop the underlying thread.
+        terminate();
     }
 
     os::ThreadInterface* Activity::thread() {
@@ -142,7 +146,10 @@ namespace RTT
         if ( ! Thread::isActive() )
             return false;
         //a trigger is always allowed when active
-        msg_cond.broadcast();
+        {
+            os::MutexLock lock(msg_lock);
+            msg_cond.broadcast();
+        }
         Thread::start();
         return true;
     }
@@ -158,7 +165,6 @@ namespace RTT
             return false;
         }
         mtimeout = true;
-        msg_cond.broadcast();
         Thread::start();
         return true;
     }
diff --git a/rtt/ExecutionEngine.cpp b/rtt/ExecutionEngine.cpp
index d80b63ce4..108fc9a4d 100644
--- a/rtt/ExecutionEngine.cpp
+++ b/rtt/ExecutionEngine.cpp
@@ -107,6 +107,7 @@ namespace RTT
             assert(foo);
             if ( foo->execute() == false ){
                 foo->unloaded();
+                os::MutexLock lock(msg_lock);
                 msg_cond.broadcast(); // required for waitForFunctions() (3rd party thread)
             } else {
                 f_queue->enqueue( foo );
@@ -211,16 +212,16 @@ namespace RTT
                 assert( com );
                 com->executeAndDispose();
             }
-            // there's no need to hold the lock during
-            // emptying the queue. But we must hold the
-            // lock once between excuteAndDispose and the
-            // broadcast to avoid the race condition in
-            // waitForMessages().
-            // This allows us to recurse into processMessages.
-            MutexLock locker( msg_lock );
         }
-        if ( com )
-            msg_cond.broadcast(); // required for waitForMessages() (3rd party thread)
+
+        // there's no need to hold the lock during
+        // emptying the queue. But we must hold the
+        // lock once between excuteAndDispose and the
+        // broadcast to avoid the race condition in
+        // waitForMessages().
+        // This allows us to recurse into processMessages.
+        MutexLock locker( msg_lock );
+        msg_cond.broadcast(); // required for waitForMessages() (3rd party thread)
     }
 
     bool ExecutionEngine::process( DisposableInterface* c )
@@ -237,7 +238,10 @@ namespace RTT
 
             bool result = mqueue->enqueue( c );
             this->getActivity()->trigger();
-            msg_cond.broadcast(); // required for waitAndProcessMessages() (EE thread)
+            {
+                os::MutexLock lock(msg_lock);
+                msg_cond.broadcast(); // required for waitAndProcessMessages() (EE thread)
+            }
             return result;
         }
         return false;
diff --git a/rtt/os/Thread.cpp b/rtt/os/Thread.cpp
index a427146a9..0c6618472 100644
--- a/rtt/os/Thread.cpp
+++ b/rtt/os/Thread.cpp
@@ -626,6 +626,7 @@ namespace RTT {
             rtos_sem_signal(&sem);
 
             rtos_task_delete(&rtos_task); // this must join the thread.
+            active = false;
         }
 
         const char* Thread::getName() const