Changeset 425 for experimental/Threading

Timestamp:

Oct 21, 2012, 12:05:47 AM (12 years ago)

Author:

Torben Dannhauer

Message:

 

Location:

experimental/Threading/Threading

Files:

• IGConnector.cpp (added)
• IGConnector.h (added)
• SimHost.cpp (modified) (3 diffs)
• SimHost.h (modified) (2 diffs)
• ThreadWorkerBase.h (modified) (3 diffs)
• ThreadedWorker.cpp (modified) (3 diffs)
• ThreadedWorker.h (modified) (2 diffs)
• Threading.vcproj (modified) (2 diffs)
• main.cpp (modified) (1 diff)

experimental/Threading/Threading/SimHost.cpp

@@ -33 +33 @@
         OpenThreads::Barrier syncBarrier;
 
-        // Create ChannelWorkers
-        OSG_ALWAYS<<"Creating ChannelWorkers..."<<std::endl;
+        // Create IG Connectors
+        OSG_ALWAYS<<"Creating IG Connectors..."<<std::endl;
         for(int i=0; i<numThreads-1; ++i)
         {
-                osg::ref_ptr<ChannelWorker> worker = new ChannelWorker();
-                _channelWorker.push_back(worker);
+                osg::ref_ptr<ThreadedWorker> threadObject = new ThreadedWorker();
+                osg::ref_ptr<IGConnector> IGCon = new IGConnector();
+                threadObject->setThreadWorker(IGCon);
+                _threadObjects.push_back(threadObject);
+                _IGConnectors.push_back(IGCon);
         }
         OSG_ALWAYS<<"...done."<<std::endl;
@@ -55 +58 @@
 {
         // Launch threads
-    for(int i=0;i<static_cast<int>(_channelWorker.size());++i)
+    for(int i=0;i<static_cast<int>(_threadObjects.size());++i)
         {
                 int status;
                 
                 //thread->setStackSize(1024*256);
-                status = _channelWorker[i]->start();
+                status = _threadObjects[i]->start();
                 assert(status == 0);
         }
 
         // Do something in the main thread
-        for(int i=0;i<10000000;i++)
-                OSG_ALWAYS<<"";
+                for(int i=0;i<10000000;i++)     OSG_ALWAYS<<"";
         
         // Signal threads to finish
-    for(int i=0;i<static_cast<int>(_channelWorker.size());++i)
+    for(int i=0;i<static_cast<int>(_threadObjects.size());++i)
         {
-                //_channelWorker[i]->setDone(true);     // non blocking
-                _channelWorker[i]->cancel();    // Blocking
+                //_threadObjects[i]->setDone(true);     // non blocking
+                _threadObjects[i]->cancel();    // Blocking
         }
 
@@ -78 +80 @@
                 OSG_ALWAYS<<"";
 
-        //syncBarrier.block(numThreads);  // Block until all threads are ready
 }

experimental/Threading/Threading/SimHost.h

@@ -5 +5 @@
 #include <osg/ref_ptr>
 
-#include "ChannelWorker.h"
+#include "ThreadedWorker.h"
+#include "IGConnector.h"
 
 
@@ -18 +19 @@
 
 private:
-        std::vector<osg::ref_ptr<ChannelWorker> > _channelWorker;
+        std::vector<osg::ref_ptr<IGConnector> > _IGConnectors;
+        std::vector<osg::ref_ptr<ThreadedWorker> > _threadObjects;
 
 };

experimental/Threading/Threading/ThreadWorkerBase.h

@@ -1 +1 @@
 #pragma once
+#include <osg/Referenced>
+
+
+class ThreadedWorker;
 
 /**
@@ -5 +9 @@
  * 
  * Just overload the operator() with your code you want to execute threaded in a loop.
- * Each execution will be completed by the thread, if the thread is cancels during a run, the cancelation will be postponed until the execution  of the functor is completed.
+ * Each execution will be completed by the thread, if the thread is set as done/canceled during a run, the cancelation will be postponed until the execution of the functor is completed.
  * 
- * Note: This functor will be executed in a loop until the thread is canceled or the parameter done is set to true.
+ * Note: This functor will be executed in a loop until the thread is canceled or set as done=true.
  * 
  * @author Torben Dannhauer
  * @date  Okt 2012
  */ 
-class ThreadWorkerBase
+class ThreadWorkerBase : public osg::Referenced
 {
 public:
@@ -18 +22 @@
         virtual ~ThreadWorkerBase() ;
 
-        void operator() (bool& _threadDone);
+        virtual void operator() (ThreadedWorker* threadObject) = 0;
+
+
+
 };

experimental/Threading/Threading/ThreadedWorker.cpp

@@ -1 +1 @@
 #include "ThreadedWorker.h"
 #include <osg/Notify>
+#include <OpenThreads/ScopedLock>
 
 
@@ -19 +20 @@
 
         // Do the work
-        int i=0;
         do
         {
-                OSG_ALWAYS<<"ID "<<getThreadId()<<" :: itr "<<i++<<std::endl;
+                OpenThreads::ScopedLock<OpenThreads::Mutex> lock(_threadWorkerMutex);
 
-                // Place Holder
-                for(int i=0;i<100;i++)
-                OSG_ALWAYS<<"";
-
-        } while (!testCancel() && !_done);      // repeat as long it is canceld or set as Done
+                if( _threadWorker.valid() )
+                        (*_threadWorker)(this);
+                
+        } while (_threadWorker!=NULL && !testCancel() && !_done);       // repeat as long as a functor is set and it is not canceled or set as Done
 }
 
@@ -60 +59 @@
     return 0;
 }
+
+void ThreadedWorker::setThreadWorker( ThreadWorkerBase* tworker)
+{
+        OpenThreads::ScopedLock<OpenThreads::Mutex> lock(_threadWorkerMutex);
+
+        _threadWorker = tworker;
+}

experimental/Threading/Threading/ThreadedWorker.h

@@ -2 +2 @@
 
 #include <osg/Referenced>
+#include <osg/ref_ptr>
 #include <OpenThreads/Thread>
+#include <OpenThreads/Mutex>
+
+#include "ThreadWorkerBase.h"
+
 
 /**
  * \brief This class is a Thread Object which allows to create Functors and attach them to this thread object.
+ * 
+ * Subclass ThreadWorkerBase to implement your own functor to execute your custom code threaded.
+ * 
+ * Use setDone(true) to shut down the thread nonblocking, or call cancel() to shut down the thread blocking (both called from outside of the workerthread).
+ * Inside the workerthread's functor, you can set the functor parameter done=true to shutdown the thread after the current execution.
+ *
  * 
  * @author Torben Dannhauer
@@ -18 +29 @@
         virtual int cancel();
         
-        void setDone(bool done);        // sinbgals the thread to stop but returns immediately. Use cancel if you want to wait blocking until the thread is canceled.
+        void setDone(bool done);        // signals the thread to stop but returns immediately. Use cancel if you want to wait blocking until the thread is canceled.
         bool getDone() const { return _done; }  
+
+        void setThreadWorker( ThreadWorkerBase* tworker);       // This function sets the functor which should be executed threaded. Set NULL to remove the current functor and cancel the thread.
+        ThreadWorkerBase* getCurrentThreadWorker()              {return(_threadWorker);};
 
 private:
         bool _done;
+        osg::ref_ptr<ThreadWorkerBase> _threadWorker;
+        OpenThreads::Mutex _threadWorkerMutex;  // Ensures the functor is not replacable while it is executed
 };

experimental/Threading/Threading/Threading.vcproj

@@ -182 +182 @@
                         >
                         <File
+                                RelativePath=".\IGConnector.cpp"
+                                >
+                        </File>
+                        <File
                                 RelativePath=".\main.cpp"
                                 >
@@ -204 +208 @@
                         >
                         <File
+                                RelativePath=".\IGConnector.h"
+                                >
+                        </File>
+                        <File
                                 RelativePath=".\SimHost.h"
                                 >

experimental/Threading/Threading/main.cpp

@@ -9 +9 @@
 int main( int argc, char **argv )
 {
-
         // Create osgVisual::SimHost
-        OSG_ALWAYS<<"Creating SimHost..."<<std::endl;
         osg::ref_ptr<SimHost> simulationHost = new SimHost();
-        OSG_ALWAYS<<"...done."<<std::endl;
 
         //Start it