Pipeline and Pipefilters can be reset for re-use

BlueBrain · Mar 4, 2016 · 6db9e3f · 6db9e3f
1 parent 8e24893
commit 6db9e3f
Show file tree

Hide file tree

Showing 13 changed files with 158 additions and 18 deletions.
diff --git a/livre/core/pipeline/Executable.h b/livre/core/pipeline/Executable.h
@@ -57,6 +57,11 @@ class Executable
      */
     virtual Futures getPreconditions() const { return Futures(); }
 
+    /**
+     * Resets the executable
+     */
+    virtual void reset() {}
+
 };
 
 }

diff --git a/livre/core/pipeline/Future.h b/livre/core/pipeline/Future.h
@@ -100,6 +100,12 @@ class Future
      */
     const AsyncData& getAsyncData() const;
 
+    /**
+     * @param future is the future to be checked with
+     * @return true if both futures are same
+     */
+    bool operator==( const Future& future ) const { return _impl == future._impl; }
+
 private:
 
     friend bool livre::waitForAny( const Futures& future );

diff --git a/livre/core/pipeline/FutureMap.cpp b/livre/core/pipeline/FutureMap.cpp
@@ -47,7 +47,6 @@ struct FutureMapImpl
         if( portName == ALL_PORTS )
         {
             Futures futures;
-            futures.reserve( _futureMap.size( ));
             for( const auto& pair: _futureMap )
                 futures.push_back( pair.second );
             return futures;
@@ -163,7 +162,7 @@ Futures OutFutureMap::getFutures() const
 
 Future OutFutureMap::getFuture( const std::string& portName ) const
 {
-    return _impl->getFutures( portName )[ 0 ];
+    return _impl->getFutures( portName ).front();
 }
 
 bool OutFutureMap::isReady( const std::string& portName ) const

diff --git a/livre/core/pipeline/InputPort.cpp b/livre/core/pipeline/InputPort.cpp
@@ -57,6 +57,20 @@ struct InputPort::Impl
         _futures.push_back( outputPort.getPromise()->getFuture( ));
     }
 
+    void disconnect( const OutputPort& outputPort )
+    {
+        Futures::iterator it = _futures.begin();
+        while( it != _futures.end())
+        {
+            if( *it == outputPort.getPromise()->getFuture( ))
+            {
+                _futures.erase( it );
+                return;
+            }
+            ++it;
+        }
+    }
+
     Futures _futures;
     const PortInfo _info;
 };
@@ -78,6 +92,11 @@ void InputPort::connect( const OutputPort& outputPort )
     _impl->connect( outputPort );
 }
 
+void InputPort::disconnect( const OutputPort& outputPort )
+{
+    _impl->disconnect( outputPort );
+}
+
 const std::string& InputPort::getName() const
 {
     return _impl->getName();

diff --git a/livre/core/pipeline/InputPort.h b/livre/core/pipeline/InputPort.h
@@ -68,6 +68,12 @@ class InputPort
      */
     void connect( const OutputPort& inputPort );
 
+    /**
+     * Disconnects an output port from input port
+     * @param inputPort input port
+     */
+    void disconnect( const OutputPort& inputPort );
+
 private:
 
     struct Impl;

diff --git a/livre/core/pipeline/OutputPort.cpp b/livre/core/pipeline/OutputPort.cpp
@@ -37,7 +37,10 @@ namespace
        Therefore, the get/set/query operations on promises/futures causes deadlocks
        when used with wait_for_any( futurelist ) if futurelist is including the
        future to be queried. With below implementation no locking is needed between
-       future operations. A higher granularity can be added for checking whet */
+       future operations. A higher granularity can be added for checking whether
+       the list includes the future or not, but simply exiting the wait_for_any
+       operation when lock is owned is simpler.
+     */
 
     ReadWriteMutex waitForAnyLock;
 }
@@ -88,6 +91,13 @@ struct AsyncData::Impl
         _future.wait();
     }
 
+    void reset()
+    {
+        ConstPortDataPromise newPromise;
+        _promise.swap( newPromise );
+        _future = _promise.get_future();
+    }
+
     ConstPortDataPromise _promise;
     mutable ConstPortDataFuture _future;
     const std::string _name;
@@ -124,6 +134,10 @@ void AsyncData::wait() const
     _impl->wait();
 }
 
+void AsyncData::reset()
+{
+    _impl->reset();
+}
 
 struct OutputPort::Impl
 {
@@ -148,6 +162,11 @@ struct OutputPort::Impl
         return _info.name;
     }
 
+    void reset()
+    {
+        _data.reset();
+    }
+
     const PortInfo _info;
     AsyncData _data;
     PromisePtr _portPromise;
@@ -181,6 +200,11 @@ void OutputPort::connect( InputPort& inputPort )
     inputPort.connect( *this );
 }
 
+void OutputPort::reset()
+{
+    _impl->reset();
+}
+
 bool waitForAny( const Futures& futures )
 {
     if( futures.empty( ))

diff --git a/livre/core/pipeline/OutputPort.h b/livre/core/pipeline/OutputPort.h
@@ -67,6 +67,11 @@ class AsyncData
      */
     void wait() const;
 
+    /**
+     * Resets the promise/future
+     */
+    void reset();
+
 private:
 
     /**
@@ -113,6 +118,11 @@ class OutputPort
      */
     void connect( InputPort& inputPort );
 
+    /**
+     * Resets the promise/future
+     */
+    void reset();
+
 private:
 
     struct Impl;

diff --git a/livre/core/pipeline/PipeFilter.cpp b/livre/core/pipeline/PipeFilter.cpp
@@ -157,7 +157,6 @@ struct PipeFilter::Impl
     Futures getPostconditions() const
     {
         Futures futures;
-        futures.reserve(_outputMap.size( ));
         for( const auto& pair: _outputMap )
         {
             const Future& outputFuture = pair.second->getPromise()->getFuture();
@@ -203,6 +202,16 @@ struct PipeFilter::Impl
         inputPort->connect( *outputPort );
     }
 
+    void reset()
+    {
+        for( auto& pair: _readyPortsMap )
+            _inputMap[ pair.first ]->disconnect( *pair.second );
+
+        _readyPortsMap.clear();
+        for( const auto& pair: _outputMap )
+            pair.second->reset();
+    }
+
     PipeFilter& _pipeFilter;
     const std::string _name;
     const servus::uint128_t _id;
@@ -241,6 +250,11 @@ const servus::uint128_t& PipeFilter::getId() const
     return _impl->_id;
 }
 
+void PipeFilter::reset()
+{
+    _impl->reset();
+}
+
 const std::string& PipeFilter::getName() const
 {
     return _impl->_name;

diff --git a/livre/core/pipeline/PipeFilter.h b/livre/core/pipeline/PipeFilter.h
@@ -118,6 +118,11 @@ class PipeFilter : public Executable
      */
     const servus::uint128_t& getId() const;
 
+    /**
+     * Resets the filter. At this point pipe filter execution should be complete.
+     */
+    void reset();
+
 private:
 
     friend class Pipeline;

diff --git a/livre/core/pipeline/Pipeline.cpp b/livre/core/pipeline/Pipeline.cpp
@@ -132,6 +132,12 @@ struct Pipeline::Impl
         return _waitFutures;
     }
 
+    void reset()
+    {
+        for( auto pair: _executableMap )
+            pair.second->reset();
+    }
+
     ~Impl()
     {}
 
@@ -185,6 +191,11 @@ void Pipeline::execute()
     return _impl->execute();
 }
 
+void Pipeline::reset()
+{
+    _impl->reset();
+}
+
 Futures Pipeline::getPreconditions() const
 {
     return _impl->getConnectedInFutureMap();

diff --git a/livre/core/pipeline/Pipeline.h b/livre/core/pipeline/Pipeline.h
@@ -93,6 +93,11 @@ class Pipeline : public Executable
      */
     void execute() final;
 
+    /**
+     * Resets the pipeline. At this point pipeline execution should be complete.
+     */
+    void reset() final;
+
 private:
 
     /**

diff --git a/livre/core/types.h b/livre/core/types.h
@@ -240,9 +240,9 @@ typedef std::vector< ConstOutputPortPtr > ConstOutputPorts;
 typedef std::vector< ConstInputPortPtr > ConstInputPorts;
 typedef std::vector< NameTypePair > NameTypePairs;
 typedef std::vector< PortInfo > PortInfos;
-typedef std::vector< Future > Futures;
 typedef std::vector< PromisePtr > Promises;
 
+typedef std::list< Future > Futures;
 typedef std::list< ExecutablePtr > Executables;
 
 template <class T>

diff --git a/tests/pipeline/pipeline.cpp b/tests/pipeline/pipeline.cpp
@@ -198,13 +198,14 @@ BOOST_AUTO_TEST_CASE( testConnection )
     BOOST_CHECK_EQUAL( outputData.thanksForAllTheFish, 222 );
 }
 
-livre::PipelinePtr createPipeline( livre::PipeFilterPtr& pipeOutput,
+livre::PipelinePtr createPipeline( livre::PipeFilterPtr& pipeInput,
+                                   livre::PipeFilterPtr& pipeOutput,
                                    uint32_t inputValue,
                                    size_t nConvertFilter = 1 )
 {
     livre::FilterPtr filter( new TestFilter( ));
     livre::PipelinePtr pipeline( new livre::Pipeline( ));
-    livre::PipeFilterPtr pipeInput = pipeline->add( "Producer", filter );
+    pipeInput = pipeline->add( "Producer", filter );
     pipeOutput = pipeline->add( "Consumer", filter );
 
     livre::FilterPtr convertFilter( new ConvertFilter( ));
@@ -229,9 +230,15 @@ livre::ExecutorPtr createExecutor(const size_t nbOfWorkerThreads )
 
 BOOST_AUTO_TEST_CASE( testSynchronousPipeline )
 {
-    livre::PipeFilterPtr pipeOutput;
     const uint32_t inputValue = 90;
-    livre::PipelinePtr pipeline = createPipeline( pipeOutput, inputValue );
+    livre::PipeFilterPtr pipeInput;
+    livre::PipeFilterPtr pipeOutput;
+
+    livre::PipelinePtr pipeline = createPipeline( pipeInput,
+                                                  pipeOutput,
+                                                  inputValue,
+                                                  1 );
+
     pipeline->execute();
 
     const livre::OutFutureMap portFutures( pipeOutput->getPostconditions( ));
@@ -241,9 +248,15 @@ BOOST_AUTO_TEST_CASE( testSynchronousPipeline )
 
 BOOST_AUTO_TEST_CASE( testWaitPipeline )
 {
-    livre::PipeFilterPtr pipeOutput;
     const uint32_t inputValue = 90;
-    livre::PipelinePtr pipeline = createPipeline( pipeOutput, inputValue );
+    livre::PipeFilterPtr pipeInput;
+    livre::PipeFilterPtr pipeOutput;
+
+    livre::PipelinePtr pipeline = createPipeline( pipeInput,
+                                                  pipeOutput,
+                                                  inputValue,
+                                                  1 );
+
     livre::ExecutorPtr executor = createExecutor( 2 );
 
     const livre::FutureMap pipelineFutures( executor->execute( pipeline->getExecutables( )));
@@ -256,9 +269,15 @@ BOOST_AUTO_TEST_CASE( testWaitPipeline )
 
 BOOST_AUTO_TEST_CASE( testAsynchronousPipeline )
 {
-    livre::PipeFilterPtr pipeOutput;
     const uint32_t inputValue = 90;
-    livre::PipelinePtr pipeline = createPipeline( pipeOutput, inputValue );
+    livre::PipeFilterPtr pipeInput;
+    livre::PipeFilterPtr pipeOutput;
+
+    livre::PipelinePtr pipeline = createPipeline( pipeInput,
+                                                  pipeOutput,
+                                                  inputValue,
+                                                  1 );
+
     livre::ExecutorPtr executor = createExecutor( 2 );
 
     executor->execute( pipeline->getExecutables( ));
@@ -274,11 +293,14 @@ BOOST_AUTO_TEST_CASE( testOneToManyManyToOnePipeline )
     {
         const size_t convertFilterCount = 10;
         const uint32_t inputValue = 90;
+        livre::PipeFilterPtr pipeInput;
         livre::PipeFilterPtr pipeOutput;
 
-        livre::PipelinePtr pipeline = createPipeline( pipeOutput,
+        livre::PipelinePtr pipeline = createPipeline( pipeInput,
+                                                      pipeOutput,
                                                       inputValue,
                                                       convertFilterCount );
+
         livre::ExecutorPtr executor = createExecutor( 1 );
         executor->execute( pipeline->getExecutables( ));
 
@@ -291,17 +313,31 @@ BOOST_AUTO_TEST_CASE( testOneToManyManyToOnePipeline )
     {
         const size_t convertFilterCount = 10;
         const uint32_t inputValue = 90;
+        livre::PipeFilterPtr pipeInput;
         livre::PipeFilterPtr pipeOutput;
 
-        livre::PipelinePtr pipeline = createPipeline( pipeOutput,
+        livre::PipelinePtr pipeline = createPipeline( pipeInput,
+                                                      pipeOutput,
                                                       inputValue,
                                                       convertFilterCount );
         livre::ExecutorPtr executor = createExecutor( 8 );
+        const livre::Futures& futures = executor->execute( pipeline->getExecutables( ));
+
+        const livre::OutFutureMap portFutures1( pipeOutput->getPostconditions( ));
+        const OutputData& outputData1 = portFutures1.get< OutputData >( "TestOutputData" );
+        BOOST_CHECK_EQUAL( outputData1.thanksForAllTheFish, 1761 );
+
+        // Reset the pipeline but wait pipeline execution before
+        const livre::FutureMap futureMap( futures );
+        futureMap.wait();
+
+        pipeline->reset();
         executor->execute( pipeline->getExecutables( ));
+        pipeInput->getPromise( "TestInputData" )->set( InputData( inputValue ));
 
-        const livre::OutFutureMap portFutures( pipeOutput->getPostconditions( ));
-        const OutputData& outputData = portFutures.get< OutputData >( "TestOutputData" );
-        BOOST_CHECK_EQUAL( outputData.thanksForAllTheFish, 1761 );
+        const livre::OutFutureMap portFutures2( pipeOutput->getPostconditions( ));
+        const OutputData& outputData2 = portFutures2.get< OutputData >( "TestOutputData" );
+        BOOST_CHECK_EQUAL( outputData2.thanksForAllTheFish, 1761 );
     }