SlaveActivity without master ExecutionEngine (pull request)

rtt/ExecutionEngine.cpp

@@ -45,6 +45,7 @@
 #include "internal/MWSRQueue.hpp"
 #include "TaskContext.hpp"
 #include "internal/CatchConfig.hpp"
+#include "extras/SlaveActivity.hpp"
 
 #include <boost/bind.hpp>
 #include <boost/ref.hpp>
@@ -71,7 +72,8 @@ namespace RTT
     ExecutionEngine::ExecutionEngine( TaskCore* owner )
         : taskc(owner),
           mqueue(new MWSRQueue<DisposableInterface*>(ORONUM_EE_MQUEUE_SIZE) ),
-          f_queue( new MWSRQueue<ExecutableInterface*>(ORONUM_EE_MQUEUE_SIZE) )
+          f_queue( new MWSRQueue<ExecutableInterface*>(ORONUM_EE_MQUEUE_SIZE) ),
+          mmaster(0)
     {
     }
 
@@ -238,23 +240,20 @@ namespace RTT
             msg_cond.broadcast(); // required for waitForMessages() (3rd party thread)
     }
 
-    void ExecutionEngine::takeoverMessages( ExecutionEngine *remote )
-    {
-        DisposableInterface* com(0);
-        {
-            while ( remote->mqueue->dequeue( com ) ) {
-                assert( com );
-                mqueue->enqueue( com );
-            }
-        }
-    }
-
     bool ExecutionEngine::process( DisposableInterface* c )
     {
         if ( c && this->getActivity() ) {
             // We only reject running functions when we're in the FatalError state.
             if (taskc && taskc->mTaskState == TaskCore::FatalError )
                 return false;
+
+            // forward message to master ExecutionEngine if available
+            if (mmaster) {
+                bool result = mmaster->process(c);
+                msg_cond.broadcast();
+                return result;
+            }
+
             bool result = mqueue->enqueue( c );
             this->getActivity()->trigger();
             msg_cond.broadcast(); // required for waitAndProcessMessages() (EE thread)
@@ -280,6 +279,22 @@ namespace RTT
             waitForMessagesInternal(pred); // same as for messages.
     }
 
+    void ExecutionEngine::setMaster(ExecutionEngine *master)
+    {
+        mmaster = master;
+    }
+
+    void ExecutionEngine::setActivity( base::ActivityInterface* task )
+    {
+        extras::SlaveActivity *slave_activity = dynamic_cast<extras::SlaveActivity *>(task);
+        if (slave_activity) {
+            ExecutionEngine *master = dynamic_cast<ExecutionEngine *>(slave_activity->getMaster());
+            setMaster(master);
+        } else {
+            setMaster(0);
+        }
+        RTT::base::RunnableInterface::setActivity(task);
+    }
 
     void ExecutionEngine::waitForMessagesInternal(boost::function<bool(void)> const& pred)
     {

rtt/ExecutionEngine.hpp

@@ -184,10 +184,20 @@ namespace RTT
         void setExceptionTask();
 
         /**
-         * this function is mentioned for activietes that are not running by her own
-         * but need to push messages to this execution engine, like SlaveActivity
+         * Set the master ExecutionEngine.
+         * If set, all incoming messages are forwarded to the master.
+         *
+         * @param master The new master ExecutionEngine.
+         */
+        void setMaster(ExecutionEngine *master);
+
+        /**
+         * Overwritten version of RTT::base::RunnableInterface::setActivity().
+         * This version will also set the master ExecutionEngine if the new activity is a SlaveActivity that runs an ExecutionEngine.
+         *
+         * @param task The ActivityInterface running this interface.
          */
-        void takeoverMessages(ExecutionEngine *remote);
+        virtual void setActivity( base::ActivityInterface* task );
 
     protected:
         /**
@@ -253,6 +263,12 @@ namespace RTT
         os::Mutex msg_lock;
         os::Condition msg_cond;
 
+        /**
+         * A master ExecutionEngine which should process our messages.
+         * This is used for ExecutionEngines running in a SlaveActivity which forward incoming messages to their master engine.
+         */
+        ExecutionEngine *mmaster;
+
         void processMessages();
         void processFunctions();
         void processChildren();

rtt/base/ActivityInterface.cpp

@@ -67,8 +67,4 @@ bool ActivityInterface::run(RunnableInterface* _r)
     return true;
 }
 
-RunnableInterface* ActivityInterface::getRunner()
-{
-    return runner;
-}
 

rtt/base/ActivityInterface.hpp

@@ -216,12 +216,6 @@ namespace RTT
          * run this activity. Will not be null.
          */
         virtual os::ThreadInterface* thread() = 0;
-
-        /**
-         * Returns the current RunnableInterface for this
-         */
-        virtual RunnableInterface* getRunner();
-
     };
 
 }}

rtt/extras/SlaveActivity.cpp

@@ -39,7 +39,6 @@
 #include "SlaveActivity.hpp"
 #include "../os/MainThread.hpp"
 #include "Logger.hpp"
-#include <rtt/ExecutionEngine.hpp>
 
 namespace RTT {
     using namespace extras;
@@ -95,6 +94,11 @@ namespace RTT {
         return mmaster ? mmaster->thread() : os::MainThread::Instance();
     }
 
+    base::ActivityInterface *SlaveActivity::getMaster() const
+    {
+        return mmaster;
+    }
+
     bool SlaveActivity::initialize()
     {
         return true;
@@ -179,22 +183,8 @@ namespace RTT {
 
     bool SlaveActivity::trigger()
     {
-        ExecutionEngine *master = 0;
         if (mmaster)
-            master = dynamic_cast<ExecutionEngine*>( mmaster->getRunner() );
-        ExecutionEngine * r= dynamic_cast<ExecutionEngine*>( runner );
-        if(!master || !r){
-            Logger::log() << Logger::Fatal << " SlaveActivity: cannot push messages to another engine, current engine is unsupported." << Logger::endl;
-            return false;
-        }else{
-            master->takeoverMessages( r );
-        }
-
-        if(mmaster)
-        {
             return mmaster->trigger();
-        }
-
         return false;
     }
 

rtt/extras/SlaveActivity.hpp

@@ -128,6 +128,8 @@ namespace RTT
 
         os::ThreadInterface* thread();
 
+        base::ActivityInterface *getMaster() const;
+
         bool initialize();
         void step();
         void loop();