Merge pull request GafferHQ#4213 from johnhaddon/parallelProcessLocationsImprovements

SceneAlgo/FilterResults improvements

Author: johnhaddon

Changes.md

@@ -26,6 +26,8 @@ Improvements
 - Context : Optimized `hash()` method, and reduced overhead in `EditableScope`.
 - NameSwitch/Spreadsheet : Rows with an empty name are now treated as if they were disabled. See Breaking Changes for further details.
 - ContextVariables : Improved performance by around 50%.
+- FilterResults : Improved performance.
+- SceneAlgo : Reduced threading overhead for `parallelProcessLocations()`, `parallelTraverse()` and `filteredParallelTraverse()`. This is particularly noticeable when visiting locations with many children.
 
 Fixes
 -----
@@ -53,6 +55,7 @@ API
   - Added `setHoverPositionVisible()` and `getHoverPositionVisible()` accessors to control an optional position indicator drawn under the pointer.
 - Expression : Added `Engine::executeCachePolicy()` method which must be implemented by subclasses.
 - ImageAlgo : Added constants for the default channel names - `channelNameR` etc.
+- SceneAlgo : Added optional `root` argument to `filteredParallelTraverse( scene, pathMatcher )`.
 
 Breaking Changes
 ----------------
@@ -71,7 +74,9 @@ Breaking Changes
   - Removed `setPositionIncrement()/getPositionIncrement()` from `Slider`. Use `setIncrement()/getIncrement()` instead.
   - Replaced `_drawPosition()` method with `_drawValue()`.
 - StandardOptions : Removed `cameraBlur` plug. This never functioned as advertised, as the regular `transformBlur` and `deformationBlur` blur settings were applied to cameras instead. As before, a StandardAttributes node may be used to customise blur for individual cameras.
-- SceneAlgo : Changed signature of the following methods to use `GafferScene::FilterPlug` : `matchingPaths`, `filteredParallelTraverse`, `Detail::ThreadableFilteredFunctor`.
+- SceneAlgo :
+  - Changed signature of the following methods to use `GafferScene::FilterPlug` : `matchingPaths`, `filteredParallelTraverse`, `Detail::ThreadableFilteredFunctor`.
+  - Removed `filteredParallelTraverse()` overload which accepted a `Filter *`. Pass `filter->outPlug()` instead.
 - DeleteFaces / DeletePoints / DeleteCurves : The PrimitiveVariable name is now taken verbatim, rather than stripping whitespace.
 - Serialisation :
   - Disabled copy construction.

include/GafferScene/SceneAlgo.h

@@ -121,36 +121,20 @@ GAFFERSCENE_API IECore::MurmurHash matchingPathsHash( const IECore::PathMatcher
 /// };
 /// ```
 template <class ThreadableFunctor>
-void parallelProcessLocations( const GafferScene::ScenePlug *scene, ThreadableFunctor &f );
-/// As above, but starting the traversal at the specified root.
-/// \todo Add default value for `root` and remove overload above.
-template <class ThreadableFunctor>
-void parallelProcessLocations( const GafferScene::ScenePlug *scene, ThreadableFunctor &f, const ScenePlug::ScenePath &root );
+void parallelProcessLocations( const GafferScene::ScenePlug *scene, ThreadableFunctor &f, const ScenePlug::ScenePath &root = ScenePlug::ScenePath() );
 
-/// Calls a functor on all paths in the scene
-/// The functor must take ( const ScenePlug*, const ScenePlug::ScenePath& ), and can return false to prune traversal
-template <class ThreadableFunctor>
-void parallelTraverse( const ScenePlug *scene, ThreadableFunctor &f );
-/// As above, but starting the traversal at `root`.
-/// \todo Add default value for `root` and remove overload above.
+/// Calls a functor on all locations in the scene. This differs from `parallelProcessLocations()` in that a single instance
+/// of the functor is used for all locations.
+/// The functor must take `( const ScenePlug *, const ScenePlug::ScenePath & )`, and can return false to prune traversal.
 template <class ThreadableFunctor>
-void parallelTraverse( const ScenePlug *scene, ThreadableFunctor &f, const ScenePlug::ScenePath &root );
+void parallelTraverse( const ScenePlug *scene, ThreadableFunctor &f, const ScenePlug::ScenePath &root = ScenePlug::ScenePath() );
 
-/// Calls a functor on all paths in the scene that are matched by the filter.
-/// The functor must take ( const ScenePlug*, const ScenePlug::ScenePath& ), and can return false to prune traversal
-template <class ThreadableFunctor>
-void filteredParallelTraverse( const ScenePlug *scene, const GafferScene::Filter *filter, ThreadableFunctor &f );
-/// As above, but specifying the filter as a plug - typically Filter::outPlug() or
-/// FilteredSceneProcessor::filterPlug() would be passed.
-template <class ThreadableFunctor>
-void filteredParallelTraverse( const ScenePlug *scene, const FilterPlug *filterPlug, ThreadableFunctor &f );
-/// As above, but starting the traversal at `root`.
-/// \todo Add default value for `root` and remove overload above.
+/// As for `parallelTraverse()`, but only calling the functor for locations matched by the filter.
 template <class ThreadableFunctor>
-void filteredParallelTraverse( const ScenePlug *scene, const FilterPlug *filterPlug, ThreadableFunctor &f, const ScenePlug::ScenePath &root );
+void filteredParallelTraverse( const ScenePlug *scene, const FilterPlug *filterPlug, ThreadableFunctor &f, const ScenePlug::ScenePath &root = ScenePlug::ScenePath() );
 /// As above, but using a PathMatcher as a filter.
 template <class ThreadableFunctor>
-void filteredParallelTraverse( const ScenePlug *scene, const IECore::PathMatcher &filter, ThreadableFunctor &f );
+void filteredParallelTraverse( const ScenePlug *scene, const IECore::PathMatcher &filter, ThreadableFunctor &f, const ScenePlug::ScenePath &root = ScenePlug::ScenePath() );
 
 /// Globals
 /// =======

include/GafferScene/SceneAlgo.inl

@@ -36,149 +36,55 @@
 
 #include "Gaffer/Context.h"
 
-#include "tbb/task.h"
+#include "tbb/parallel_for.h"
 
 namespace GafferScene
 {
 
 namespace Detail
 {
 
-template <class ThreadableFunctor>
-class TraverseTask : public tbb::task
-{
-
-	public :
-
-		TraverseTask(
-			const GafferScene::ScenePlug *scene,
-			const Gaffer::ThreadState &threadState,
-			ThreadableFunctor &f
-		)
-			:	m_scene( scene ), m_threadState( threadState ), m_f( f )
-		{
-		}
-
-		TraverseTask(
-			const GafferScene::ScenePlug *scene,
-			const Gaffer::ThreadState &threadState,
-			const ScenePlug::ScenePath &path,
-			ThreadableFunctor &f
-		)
-			:	m_scene( scene ), m_threadState( threadState ), m_f( f ), m_path( path )
-		{
-		}
-
-
-		~TraverseTask() override
-		{
-		}
-
-		task *execute() override
-		{
-			ScenePlug::PathScope pathScope( m_threadState, &m_path );
-
-			if( m_f( m_scene, m_path ) )
-			{
-				IECore::ConstInternedStringVectorDataPtr childNamesData = m_scene->childNamesPlug()->getValue();
-				const std::vector<IECore::InternedString> &childNames = childNamesData->readable();
-
-				set_ref_count( 1 + childNames.size() );
-
-				ScenePlug::ScenePath childPath = m_path;
-				childPath.push_back( IECore::InternedString() ); // space for the child name
-				for( std::vector<IECore::InternedString>::const_iterator it = childNames.begin(), eIt = childNames.end(); it != eIt; it++ )
-				{
-					childPath[m_path.size()] = *it;
-					TraverseTask *t = new( allocate_child() ) TraverseTask( *this, childPath );
-					spawn( *t );
-				}
-				wait_for_all();
-			}
-
-			return nullptr;
-		}
-
-	protected :
-
-		TraverseTask( const TraverseTask &other, const ScenePlug::ScenePath &path )
-			:	m_scene( other.m_scene ),
-			m_threadState( other.m_threadState ),
-			m_f( other.m_f ),
-			m_path( path )
-		{
-		}
-
-	private :
-
-		const GafferScene::ScenePlug *m_scene;
-		const Gaffer::ThreadState &m_threadState;
-		ThreadableFunctor &m_f;
-		GafferScene::ScenePlug::ScenePath m_path;
-
-};
-
 template<typename ThreadableFunctor>
-class LocationTask : public tbb::task
+void parallelProcessLocationsWalk( const GafferScene::ScenePlug *scene, const Gaffer::ThreadState &threadState, const ScenePlug::ScenePath &path, ThreadableFunctor &f, tbb::task_group_context &taskGroupContext )
 {
+	ScenePlug::PathScope pathScope( threadState, &path );
 
-	public :
+	if( !f( scene, path ) )
+	{
+		return;
+	}
 
-		LocationTask(
-			const GafferScene::ScenePlug *scene,
-			const Gaffer::ThreadState &threadState,
-			const ScenePlug::ScenePath &path,
-			ThreadableFunctor &f
-		)
-			:	m_scene( scene ), m_threadState( threadState ), m_path( path ), m_f( f )
-		{
-		}
+	IECore::ConstInternedStringVectorDataPtr childNamesData = scene->childNamesPlug()->getValue();
+	const std::vector<IECore::InternedString> &childNames = childNamesData->readable();
+	if( childNames.empty() )
+	{
+		return;
+	}
 
-		~LocationTask() override
-		{
-		}
+	using ChildNameRange = tbb::blocked_range<std::vector<IECore::InternedString>::const_iterator>;
+	const ChildNameRange loopRange( childNames.begin(), childNames.end() );
 
-		task *execute() override
+	auto loopBody = [&] ( const ChildNameRange &range ) {
+		ScenePlug::ScenePath childPath = path;
+		childPath.push_back( IECore::InternedString() ); // Space for the child name
+		for( auto &childName : range )
 		{
-			ScenePlug::PathScope pathScope( m_threadState, &m_path );
-
-			if( !m_f( m_scene, m_path ) )
-			{
-				return nullptr;
-			}
-
-			IECore::ConstInternedStringVectorDataPtr childNamesData = m_scene->childNamesPlug()->getValue();
-			const std::vector<IECore::InternedString> &childNames = childNamesData->readable();
-			if( childNames.empty() )
-			{
-				return nullptr;
-			}
-
-			std::vector<ThreadableFunctor> childFunctors( childNames.size(), m_f );
-
-			set_ref_count( 1 + childNames.size() );
-
-			ScenePlug::ScenePath childPath = m_path;
-			childPath.push_back( IECore::InternedString() ); // space for the child name
-			for( size_t i = 0, e = childNames.size(); i < e; ++i )
-			{
-				childPath.back() = childNames[i];
-				LocationTask *t = new( allocate_child() ) LocationTask( m_scene, m_threadState, childPath, childFunctors[i] );
-				spawn( *t );
-			}
-			wait_for_all();
-
-			return nullptr;
+			ThreadableFunctor childFunctor( f );
+			childPath.back() = childName;
+			parallelProcessLocationsWalk( scene, threadState, childPath, childFunctor, taskGroupContext );
 		}
+	};
 
-	private :
-
-		const GafferScene::ScenePlug *m_scene;
-		const Gaffer::ThreadState &m_threadState;
-		const GafferScene::ScenePlug::ScenePath m_path;
-		ThreadableFunctor &m_f;
-
-};
+	if( childNames.size() > 1 )
+	{
+		tbb::parallel_for( loopRange, loopBody, taskGroupContext );
+	}
+	else
+	{
+		// Serial execution
+		loopBody( loopRange );
+	}
+}
 
 template <class ThreadableFunctor>
 struct ThreadableFilteredFunctor
@@ -240,47 +146,23 @@ struct PathMatcherFunctor
 namespace SceneAlgo
 {
 
-template <class ThreadableFunctor>
-void parallelProcessLocations( const GafferScene::ScenePlug *scene, ThreadableFunctor &f )
-{
-	return parallelProcessLocations( scene, f, ScenePlug::ScenePath() );
-}
-
 template <class ThreadableFunctor>
 void parallelProcessLocations( const GafferScene::ScenePlug *scene, ThreadableFunctor &f, const ScenePlug::ScenePath &root )
 {
 	tbb::task_group_context taskGroupContext( tbb::task_group_context::isolated ); // Prevents outer tasks silently cancelling our tasks
-	Detail::LocationTask<ThreadableFunctor> *task = new( tbb::task::allocate_root( taskGroupContext ) ) Detail::LocationTask<ThreadableFunctor>( scene, Gaffer::ThreadState::current(), root, f );
-	tbb::task::spawn_root_and_wait( *task );
-}
-
-template <class ThreadableFunctor>
-void parallelTraverse( const GafferScene::ScenePlug *scene, ThreadableFunctor &f )
-{
-	tbb::task_group_context taskGroupContext( tbb::task_group_context::isolated ); // Prevents outer tasks silently cancelling our tasks
-	Detail::TraverseTask<ThreadableFunctor> *task = new( tbb::task::allocate_root( taskGroupContext ) ) Detail::TraverseTask<ThreadableFunctor>( scene, Gaffer::ThreadState::current(), f );
-	tbb::task::spawn_root_and_wait( *task );
+	Detail::parallelProcessLocationsWalk( scene, Gaffer::ThreadState::current(), root, f, taskGroupContext );
 }
 
 template <class ThreadableFunctor>
 void parallelTraverse( const ScenePlug *scene, ThreadableFunctor &f, const ScenePlug::ScenePath &root )
 {
-	tbb::task_group_context taskGroupContext( tbb::task_group_context::isolated ); // Prevents outer tasks silently cancelling our tasks
-	Detail::TraverseTask<ThreadableFunctor> *task = new( tbb::task::allocate_root( taskGroupContext ) ) Detail::TraverseTask<ThreadableFunctor>( scene, Gaffer::ThreadState::current(), root, f );
-	tbb::task::spawn_root_and_wait( *task );
-}
-
-template <class ThreadableFunctor>
-void filteredParallelTraverse( const GafferScene::ScenePlug *scene, const GafferScene::Filter *filter, ThreadableFunctor &f )
-{
-	filteredParallelTraverse( scene, filter->outPlug(), f );
-}
-
-template <class ThreadableFunctor>
-void filteredParallelTraverse( const GafferScene::ScenePlug *scene, const GafferScene::FilterPlug *filterPlug, ThreadableFunctor &f )
-{
-	Detail::ThreadableFilteredFunctor<ThreadableFunctor> ff( f, filterPlug );
-	parallelTraverse( scene, ff );
+	// `parallelProcessLocations()` takes a copy of the functor at each location, whereas
+	// `parallelTraverse()` is intended to use the same functor for all locations. Wrap the
+	// functor in a cheap-to-copy lambda, so that the functor itself won't be copied.
+	auto reference = [&f] ( const ScenePlug *scene, const ScenePlug::ScenePath &path ) {
+		return f( scene, path );
+	};
+	parallelProcessLocations( scene, reference, root );
 }
 
 template <class ThreadableFunctor>
@@ -291,10 +173,10 @@ void filteredParallelTraverse( const ScenePlug *scene, const GafferScene::Filter
 }
 
 template <class ThreadableFunctor>
-void filteredParallelTraverse( const ScenePlug *scene, const IECore::PathMatcher &filter, ThreadableFunctor &f )
+void filteredParallelTraverse( const ScenePlug *scene, const IECore::PathMatcher &filter, ThreadableFunctor &f, const ScenePlug::ScenePath &root )
 {
 	Detail::PathMatcherFunctor<ThreadableFunctor> ff( f, filter );
-	parallelTraverse( scene, ff );
+	parallelTraverse( scene, ff, root );
 }
 
 } // namespace SceneAlgo

python/GafferSceneTest/FilterResultsTest.py

@@ -255,7 +255,7 @@ def testRootMatchVsNoMatch( self ) :
 
 	@unittest.skipIf( GafferTest.inCI(), "Performance not relevant on CI platform" )
 	@GafferTest.TestRunner.PerformanceTestMethod()
-	def testHashPerf( self ):
+	def testHashPerformance( self ):
 
 		sphere = GafferScene.Sphere()
 
@@ -277,5 +277,29 @@ def testHashPerf( self ):
 		with GafferTest.TestRunner.PerformanceScope():
 			filterResults["out"].hash()
 
+	@unittest.skipIf( GafferTest.inCI(), "Performance not relevant on CI platform" )
+	@GafferTest.TestRunner.PerformanceTestMethod()
+	def testComputePerformance( self ):
+
+		sphere = GafferScene.Sphere()
+
+		duplicate = GafferScene.Duplicate()
+		duplicate["in"].setInput( sphere["out"] )
+		duplicate["target"].setValue( '/sphere' )
+		duplicate["copies"].setValue( 1000000 )
+
+		pathFilter = GafferScene.PathFilter()
+		pathFilter["paths"].setValue( IECore.StringVectorData( [ '...' ] ) )
+
+		filterResults = GafferScene.FilterResults()
+		filterResults["scene"].setInput( duplicate["out"] )
+		filterResults["filter"].setInput( pathFilter["out"] )
+
+		# Evaluate the root childNames beforehand to focus our timing on the compute
+		duplicate["out"].childNames( "/" )
+
+		with GafferTest.TestRunner.PerformanceScope():
+			filterResults["out"].getValue()
+
 if __name__ == "__main__":
 	unittest.main()