related to #40: Bringing the new pipeline concept introduced with r458 to AbstractPipeline

 * Pipelines evaluate themselves
 * removed the PipelineEvaluator
 * adapted all demo pipelines except OpenCLPipeline

git-svn-id: https://camplinux.in.tum.de/svn/campvis/trunk@462 bb408c1c-ae56-11e1-83d9-df6b3e0c105e

application/campvisapplication.cpp

@@ -50,7 +50,6 @@
 #include "core/tools/quadrenderer.h"
 #include "core/pipeline/abstractpipeline.h"
 #include "core/pipeline/visualizationpipeline.h"
-#include "core/pipeline/pipelineevaluator.h"
 
 namespace campvis {
 
@@ -79,9 +78,6 @@ namespace campvis {
         tgtAssert(_initialized == false, "Destructing initialized CampVisApplication, deinitialize first!");
 
         // delete everything in the right order:
-        for (std::vector<PipelineEvaluator*>::iterator it = _pipelineEvaluators.begin(); it != _pipelineEvaluators.end(); ++it) {
-            delete *it;
-        }
         for (std::vector< std::pair<VisualizationPipeline*, TumVisPainter*> >::iterator it = _visualizations.begin(); it != _visualizations.end(); ++it) {
             delete it->second;
         }
@@ -211,19 +207,9 @@ namespace campvis {
 
         _mainWindow->show();
 
-        // Start evaluator/render threads
-        for (std::vector<PipelineEvaluator*>::iterator it = _pipelineEvaluators.begin(); it != _pipelineEvaluators.end(); ++it) {
-            (*it)->start();
-        }
-
         // Start QApplication
         int toReturn = QApplication::exec();
 
-        // QApplication has returned -> Stop evaluator/render threads
-        for (std::vector<PipelineEvaluator*>::iterator it = _pipelineEvaluators.begin(); it != _pipelineEvaluators.end(); ++it) {
-            (*it)->stop();
-        }
-
         return toReturn;
     }
 
@@ -232,9 +218,6 @@ namespace campvis {
         tgtAssert(pipeline != 0, "Pipeline must not be 0.");
         _pipelines.push_back(pipeline);
 
-        PipelineEvaluator* pe = new PipelineEvaluator(pipeline);
-        _pipelineEvaluators.push_back(pe);
-
         s_PipelinesChanged();
     }
 

application/campvisapplication.h

@@ -44,7 +44,6 @@ namespace tgt {
 namespace campvis {
     class AbstractPipeline;
     class MainWindow;
-    class PipelineEvaluator;
     class TumVisPainter;
     class VisualizationPipeline;
 
@@ -122,8 +121,6 @@ namespace campvis {
     private:
         /// All pipelines (incuding VisualizationPipelines)
         std::vector<AbstractPipeline*> _pipelines;
-        /// All pipeline evaluators (separated from _pipelines because we probably want multiple pipelines per evaluator later)
-        std::vector<PipelineEvaluator*> _pipelineEvaluators;
         /// All visualisations (i.e. VisualizationPipelines with their corresponding painters/canvases)
         std::vector< std::pair<VisualizationPipeline*, TumVisPainter*> > _visualizations;
 

core/datastructures/imagedata.h

@@ -224,13 +224,12 @@ namespace campvis {
     const T* campvis::ImageData::getRepresentation(bool performConversion) const {
         // look, whether we already have a suitable representation
         for (tbb::concurrent_vector<const AbstractImageRepresentation*>::const_iterator it = _representations.begin(); it != _representations.end(); ++it) {
-            //if (const T* tester = dynamic_cast<const T*>(*it))
-            //    return tester;
             if (typeid(T) == typeid(**it)) {
                 return static_cast<const T*>(*it);
             }
         }
 
+        // no representation found, create a new one
         if (performConversion) {
             return tryPerformConversion<T>();
         }
@@ -248,7 +247,9 @@ namespace campvis {
 
     template<typename T>
     const T* campvis::ImageData::tryPerformConversion() const {
-        // no representation found, create a new one
+        // TODO: Currently, we do not check, for multiple parallel conversions into the same
+        //       target type. This does not harm thread-safety but may lead to multiple 
+        //       representations of the same type for a single image.
         for (tbb::concurrent_vector<const AbstractImageRepresentation*>::const_iterator it = _representations.begin(); it != _representations.end(); ++it) {
             const T* tester = T::tryConvertFrom(*it);
             if (tester != 0) {

core/pipeline/abstractpipeline.cpp

@@ -31,6 +31,8 @@
 
 #include "tgt/exception.h"
 #include "core/pipeline/abstractprocessor.h"
+#include "core/tools/job.h"
+#include "core/tools/simplejobprocessor.h"
 #include <ctime>
 
 
@@ -39,8 +41,8 @@ namespace campvis {
 
     AbstractPipeline::AbstractPipeline() 
         : HasPropertyCollection()
-        , _enabled(true)
     {
+        _enabled = true;
     }
 
     AbstractPipeline::~AbstractPipeline() {
@@ -78,12 +80,13 @@ namespace campvis {
 
     void AbstractPipeline::onPropertyChanged(const AbstractProperty* prop) {
         HasPropertyCollection::onPropertyChanged(prop);
-         _invalidationLevel.setLevel(InvalidationLevel::INVALID_RESULT);
         s_PipelineInvalidated();
     }
 
     void AbstractPipeline::onProcessorInvalidated(AbstractProcessor* processor) {
-        _invalidationLevel.setLevel(InvalidationLevel::INVALID_RESULT);
+        if (processor->getEnabled())
+            SimpleJobProc.enqueueJob(makeJob(this, &AbstractPipeline::executeProcessor, processor));
+
         s_PipelineInvalidated();
     }
 
@@ -119,15 +122,11 @@ namespace campvis {
             processor->unlockProcessor();
 
 #ifdef CAMPVIS_DEBUG
-            LDEBUG("Executed processor " << processor->getName() << " duration: " << (endTime - startTime));
+            //LDEBUG("Executed processor " << processor->getName() << " duration: " << (endTime - startTime));
 #endif
         }
     }
 
-    InvalidationLevel& AbstractPipeline::getInvalidationLevel() {
-        return _invalidationLevel;
-    }
-
     const std::vector<AbstractProcessor*>& AbstractPipeline::getProcessors() const {
         return _processors;
     }
@@ -145,5 +144,15 @@ namespace campvis {
         _processors.push_back(processor);
     }
 
+    void AbstractPipeline::lockAllProcessors() {
+        for (std::vector<AbstractProcessor*>::iterator it = _processors.begin(); it != _processors.end(); ++it)
+            (*it)->lockProcessor();
+    }
+
+    void AbstractPipeline::unlockAllProcessors() {
+        for (std::vector<AbstractProcessor*>::iterator it = _processors.begin(); it != _processors.end(); ++it)
+            (*it)->unlockProcessor();
+    }
+
 
 }

core/pipeline/abstractpipeline.h

@@ -77,11 +77,6 @@ namespace campvis {
          */
         virtual void deinit();
 
-        /**
-         * Execute this pipeline.
-         **/
-        virtual void execute() = 0;
-
         /**
          * Returns the DataContainer of this pipeline, const version.
          * \return _data
@@ -114,12 +109,6 @@ namespace campvis {
          * \return  The name of this pipeline.
          */
         virtual const std::string getName() const = 0;
-        
-        /**
-         * Gets the current InvalidationLevel of this pipeline.
-         * \return _invalidationLevel
-         */
-        InvalidationLevel& getInvalidationLevel();
 
         /**
          * Gets the flag whether this pipeline is currently enabled.
@@ -137,6 +126,16 @@ namespace campvis {
         sigslot::signal0<> s_PipelineInvalidated;
 
     protected:
+        /**
+         * Locks all processors.
+         */
+        void lockAllProcessors();
+
+        /**
+         * Unlocks all processors.
+         */
+        void unlockAllProcessors();
+
         /**
          * Slot getting called when one of the observed properties changed and notifies its observers.
          * The default behaviour is just to set the invalidation level to invalid.
@@ -161,11 +160,9 @@ namespace campvis {
         DataContainer _data;                                ///< DataContainer containing local working set of data for this Pipeline
 
         std::vector<AbstractProcessor*> _processors;        ///< List of all processors of this pipeline
-        InvalidationLevel _invalidationLevel;               ///< current invalidation level
-        bool _enabled;                                      ///< flag whether this pipeline is currently enabled
+        tbb::atomic<bool> _enabled;                         ///< flag whether this pipeline is currently enabled
 
-        tbb::spin_mutex _localMutex;                        ///< mutex for altering local members
-        tbb::mutex _evaluationMutex;                        ///< mutex for the evaluation of this pipeline
+        //tbb::spin_mutex _localMutex;                        ///< mutex for altering local members
 
         static const std::string loggerCat_;
     };

core/pipeline/abstractprocessor.h

@@ -172,7 +172,6 @@ namespace campvis {
 
 
     protected:
-        InvalidationLevel _invalidationLevel;       ///< current invalidation level of this processor
         tbb::atomic<bool> _enabled;                 ///< flag whether this processor is currently enabled
 
         /// Flag whether this processor is currently locked
@@ -180,6 +179,10 @@ namespace campvis {
         tbb::atomic<bool> _locked;
 
         static const std::string loggerCat_;
+
+    private:
+        InvalidationLevel _invalidationLevel;       ///< current invalidation level of this processor
+
     };
 
 }

core/pipeline/pipelineevaluator.cpp

-// ================================================================================================
-// 
-// This file is part of the CAMPVis Software Framework.
-// 
-// If not explicitly stated otherwise: Copyright (C) 2012, all rights reserved,
-//      Christian Schulte zu Berge <christian.szb@in.tum.de>
-//      Chair for Computer Aided Medical Procedures
-//      Technische Universität München
-//      Boltzmannstr. 3, 85748 Garching b. München, Germany
-// For a full list of authors and contributors, please refer to the file "AUTHORS.txt".
-// 
-// The licensing of this softare is not yet resolved. Until then, redistribution in source or
-// binary forms outside the CAMP chair is not permitted, unless explicitly stated in legal form.
-// However, the names of the original authors and the above copyright notice must retain in its
-// original state in any case.
-// 
-// Legal disclaimer provided by the BSD license:
-// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR
-// IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY 
-// AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR 
-// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
-// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 
-// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 
-// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR 
-// OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
-// POSSIBILITY OF SUCH DAMAGE.
-// 
-// ================================================================================================
-
-#include "pipelineevaluator.h"
-#include "tgt/qt/qtcontextmanager.h"
-#include "core/pipeline/abstractpipeline.h"
-
-namespace campvis {
-    PipelineEvaluator::PipelineEvaluator(AbstractPipeline* pipeline)
-        : Runnable()
-        , _pipeline(pipeline)
-    {
-        tgtAssert(pipeline != 0, "Pipeline must not be 0.");
-        pipeline->s_PipelineInvalidated.connect(this, &PipelineEvaluator::OnPipelineInvalidated);
-    }
-
-    PipelineEvaluator::~PipelineEvaluator() {
-        _pipeline->s_PipelineInvalidated.disconnect(this);
-    }
-
-    void PipelineEvaluator::run() {
-        std::unique_lock<tbb::mutex> lock(_pipeline->_evaluationMutex);
-
-        while (! _stopExecution) {
-            if (_pipeline->getEnabled())
-                _pipeline->execute();
-
-            while (!_stopExecution && _pipeline->getInvalidationLevel().isValid())
-                _evaluationCondition.wait(lock);
-        }
-
-        // release OpenGL context, so that other threads can access it
-        CtxtMgr.releaseCurrentContext();
-    }
-
-    void PipelineEvaluator::stop() {
-        // we need to execute run() one more time to ensure correct release of the OpenGL context
-        _stopExecution = true;
-        _evaluationCondition.notify_all();
-
-        Runnable::stop();
-    }
-
-    void PipelineEvaluator::OnPipelineInvalidated() {
-        if (!_stopExecution)
-            _evaluationCondition.notify_all();
-    }
-
-}

core/pipeline/pipelineevaluator.h

-// ================================================================================================
-// 
-// This file is part of the CAMPVis Software Framework.
-// 
-// If not explicitly stated otherwise: Copyright (C) 2012, all rights reserved,
-//      Christian Schulte zu Berge <christian.szb@in.tum.de>
-//      Chair for Computer Aided Medical Procedures
-//      Technische Universität München
-//      Boltzmannstr. 3, 85748 Garching b. München, Germany
-// For a full list of authors and contributors, please refer to the file "AUTHORS.txt".
-// 
-// The licensing of this softare is not yet resolved. Until then, redistribution in source or
-// binary forms outside the CAMP chair is not permitted, unless explicitly stated in legal form.
-// However, the names of the original authors and the above copyright notice must retain in its
-// original state in any case.
-// 
-// Legal disclaimer provided by the BSD license:
-// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR
-// IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY 
-// AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR 
-// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
-// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 
-// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 
-// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR 
-// OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
-// POSSIBILITY OF SUCH DAMAGE.
-// 
-// ================================================================================================
-
-#ifndef PIPELINEEVALUATOR_H__
-#define PIPELINEEVALUATOR_H__
-
-#include "sigslot/sigslot.h"
-#include "core/tools/runnable.h"
-#include "tbb/compat/condition_variable"
-
-namespace campvis {
-    class AbstractPipeline;
-
-    /**
-     * The PipelineEvaluator evaluates its pipeline in its own thread.
-     * Evaluation is implemented using condidional wait - hence the pipeline is only evaluated when
-     * \a pipeline emits the s_PipelineInvalidated signal.
-     * 
-     * \todo    Let one PipelineEvaluator evaluate multiple pipelines.
-     * \sa      Runnable
-     */
-    class PipelineEvaluator : public Runnable, public sigslot::has_slots<> {
-    public:
-        /**
-         * Creates a new PipelineEvaluator for the given pipeline \a pipeline.
-         * \param   pipeline    Pipeline to evaluate
-         */
-        PipelineEvaluator(AbstractPipeline* pipeline);
-
-        /**
-         * Destructor, stops and waits for the evaluation thread if it's still running.
-         */
-        virtual ~PipelineEvaluator();
-
-        /// \see Runnable::stop
-        void stop();
-        
-        /**
-         * Performs the pipeline evaluation using conditional wait.
-         * \sa Runnable::run
-         */
-        void run();
-
-        /**
-         * Slot for notifications when the pipeline was invalidated.
-         */
-        void OnPipelineInvalidated();
-
-    protected:
-        AbstractPipeline* _pipeline;                    ///< Pipeline to evaluate
-        std::condition_variable _evaluationCondition;   ///< conditional wait condition
-    };
-
-}
-
-#endif // PIPELINEEVALUATOR_H__

core/pipeline/raycastingprocessor.cpp

@@ -98,7 +98,7 @@ namespace campvis {
                     p_transferFunction.getTF()->setImageHandle(img.getDataHandle());
                 }
 
-                if (_invalidationLevel.isInvalidShader()) {
+                if (getInvalidationLevel().isInvalidShader()) {
                     _shader->setHeaders(generateHeader());
                     _shader->rebuild();
                 }
@@ -158,7 +158,7 @@ namespace campvis {
             LERROR("No suitable input image found.");
         }
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 
     std::string RaycastingProcessor::generateHeader() const {

core/pipeline/visualizationpipeline.cpp

@@ -32,8 +32,10 @@
 #include "tgt/glcanvas.h"
 #include "tgt/glcontext.h"
 #include "core/datastructures/imagerepresentationrendertarget.h"
+#include "core/pipeline/visualizationprocessor.h"
 #include "core/tools/job.h"
 #include "core/tools/opengljobprocessor.h"
+#include "core/tools/simplejobprocessor.h"
 
 namespace campvis {
     const std::string VisualizationPipeline::loggerCat_ = "CAMPVis.core.datastructures.VisualizationPipeline";
@@ -136,4 +138,32 @@ namespace campvis {
             _effectiveRenderTargetSize.setValue(_renderTargetSize);
     }
 
+    void VisualizationPipeline::onProcessorInvalidated(AbstractProcessor* processor) {
+        if (_canvas == 0)
+            return;
+
+        tbb::concurrent_hash_map<AbstractProcessor*, bool>::const_accessor a;
+        if (_isVisProcessorMap.find(a, processor)) {
+            if (a->second) {
+                // is VisualizationProcessor
+                GLJobProc.enqueueJob(
+                    _canvas, 
+                    makeJobOnHeap<VisualizationPipeline, AbstractProcessor*>(this, &VisualizationPipeline::executeProcessor, processor), 
+                    OpenGLJobProcessor::SerialJob);
+            }
+            else {
+                SimpleJobProc.enqueueJob(makeJob<VisualizationPipeline, AbstractProcessor*>(this, &VisualizationPipeline::executeProcessor, processor));
+            }
+        }
+        else {
+            tgtAssert(false, "Could not find processor in processor map.");
+            LWARNING("Caught invalidation of a non-registered processor!");
+        }
+    }
+
+    void VisualizationPipeline::addProcessor(AbstractProcessor* processor) {
+        _isVisProcessorMap.insert(std::make_pair(processor, (dynamic_cast<VisualizationProcessor*>(processor) != 0)));
+        AbstractPipeline::addProcessor(processor);
+    }
+
 }

core/pipeline/visualizationpipeline.h

@@ -31,6 +31,7 @@
 #define VISUALIZATIONPIPELINE_H__
 
 #include "sigslot/sigslot.h"
+#include "tbb/concurrent_hash_map.h"
 #include "tgt/vector.h"
 #include "tgt/event/eventlistener.h"
 #include "core/eventhandlers/abstracteventhandler.h"
@@ -78,10 +79,13 @@ namespace campvis {
         virtual void deinit();
 
         /**
-         * Execute this pipeline.
-         * Pipeline must have a valid canvas set before calling this method.
-         **/
-        virtual void execute() = 0;
+         * Adds the processor \a processor to this pipeline's processor list.
+         * \note    The s_invalidated signal of each processor on this list will be automatically 
+         *          connected to onProcessorInvalidated() during initialization and disconnected
+         *          during deinitialization.
+         * \param   processor   The processor to add.
+         */
+        void addProcessor(AbstractProcessor* processor);
 
         /**
          * Performs the event handling for the assigned canvas.
@@ -152,6 +156,13 @@ namespace campvis {
          * \param   prop    Property that emitted the signal
          */
         virtual void onPropertyChanged(const AbstractProperty* prop);
+        
+        /**
+         * Slot getting called when one of the observed processors got invalidated.
+         * The default behaviour is just to set the invalidation level to invalid.
+         * \param   processor   The processor that emitted the signal
+         */
+        virtual void onProcessorInvalidated(AbstractProcessor* processor);
 
         /**
          * Acquires and locks the OpenGL context, executes the processor \a processor on the pipeline's data 
@@ -165,6 +176,8 @@ namespace campvis {
          */
         void updateEffectiveRenderTargetSize();
 
+        tbb::concurrent_hash_map<AbstractProcessor*, bool> _isVisProcessorMap;
+
         tgt::ivec2 _renderTargetSize;                           ///< original render target size
         bool _lqMode;                                           ///< Flag whether low quality mode is enables
 

modules/advancedusvis/pipelines/advancedusvis.cpp

@@ -198,11 +198,6 @@ namespace campvis {
     }
 
     void AdvancedUsVis::execute() {
-        {
-            tbb::spin_mutex::scoped_lock lock(_localMutex);
-            _invalidationLevel.setValid();
-            // TODO:    think whether we want to lock all processors already here.
-        }
 /*
         if (!_usReader.getInvalidationLevel().isValid()) {
             SimpleJobProc.enqueueJob(makeJob(this, &AdvancedUsVis::foobar));

modules/advancedusvis/processors/advancedusfusion.cpp

@@ -145,7 +145,7 @@ namespace campvis {
             LERROR("No suitable input image found.");
         }
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 
     void AdvancedUsFusion::updateProperties(DataHandle img) {

modules/devil/processors/devilimagereader.cpp

@@ -124,6 +124,6 @@ namespace campvis {
             LERROR("Could not load image.");
         }
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 }
\ No newline at end of file

modules/io/processors/csvdimagereader.cpp

@@ -171,6 +171,6 @@ namespace campvis {
             return;
         }
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 }
\ No newline at end of file

modules/io/processors/mhdimagereader.cpp

@@ -187,6 +187,6 @@ namespace campvis {
             return;
         }
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 }
\ No newline at end of file

modules/io/processors/vtkimagereader.cpp

@@ -205,6 +205,6 @@ namespace campvis {
             return;
         }
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 }
\ No newline at end of file

modules/itk/processors/itkimagefilter.cpp

@@ -264,7 +264,7 @@ namespace campvis {
         }
 
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 
     void ItkImageFilter::updateProperties() {

modules/preprocessing/processors/gradientvolumegenerator.cpp

@@ -115,7 +115,7 @@ namespace campvis {
         }
 
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 
 }

modules/preprocessing/processors/imagefilter.cpp

@@ -91,7 +91,7 @@ namespace campvis {
         }
 
 
-        _invalidationLevel.setValid();
+        applyInvalidationLevel(InvalidationLevel::VALID);
     }
 
 }