AbstractMultiThreadedConvolution: shutdown automatically create ExecutorService

Author: maarzt

src/main/java/net/imglib2/algorithm/convolution/AbstractMultiThreadedConvolution.java

@@ -1,37 +1,66 @@
 package net.imglib2.algorithm.convolution;
 
+import net.imglib2.RandomAccessible;
+import net.imglib2.RandomAccessibleInterval;
+
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadPoolExecutor;
 
 /**
  * Abstract class to help implementing a Convolution, that is multi threaded
  * using an {@link ExecutorService}. This implements the method
- * {@link Convolution#setExecutor(ExecutorService)} and has useful protected
- * methods {@link #getExecutor()} and {@link #getNumThreads()}.
+ * {@link Convolution#setExecutor(ExecutorService)}.
+ * <p>
+ * Classes that derive from
+ * {@link AbstractMultiThreadedConvolution} must override
+ * {@link AbstractMultiThreadedConvolution#process(RandomAccessible, RandomAccessibleInterval, ExecutorService, int)}
  */
 public abstract class AbstractMultiThreadedConvolution< T > implements Convolution< T >
 {
-	private ExecutorService executor = null;
+
+	private ExecutorService executor;
+
+	abstract protected void process( RandomAccessible< ? extends T > source,
+			RandomAccessibleInterval< ? extends T > target,
+			ExecutorService executorService,
+			int numThreads);
 
 	@Override
 	public void setExecutor( ExecutorService executor )
 	{
 		this.executor = executor;
 	}
 
-	protected ExecutorService getExecutor()
+	@Override
+	final public void process( RandomAccessible< ? extends T > source, RandomAccessibleInterval< ? extends T > target )
 	{
-		if ( executor == null )
-			executor = Executors.newFixedThreadPool( Runtime.getRuntime().availableProcessors() );
-		return executor;
+		if(executor == null) {
+			int numThreads = suggestNumThreads();
+			ExecutorService executor = Executors.newFixedThreadPool( numThreads );
+			try
+			{
+				process( source, target, executor, numThreads );
+			}
+			finally
+			{
+				executor.shutdown();
+			}
+		}
+		else {
+			process( source, target, executor, getNumThreads( executor ) );
+		}
 	}
 
-	protected int getNumThreads()
+	private int getNumThreads(ExecutorService executor)
 	{
-		ExecutorService executor = getExecutor();
 		if ( executor instanceof ThreadPoolExecutor )
 			return ( ( ThreadPoolExecutor ) executor ).getMaximumPoolSize();
+		return suggestNumThreads();
+	}
+
+	private int suggestNumThreads()
+	{
 		return Runtime.getRuntime().availableProcessors();
 	}
 }

src/main/java/net/imglib2/algorithm/convolution/LineConvolution.java

@@ -53,21 +53,18 @@ public LineConvolution( LineConvolverFactory< ? super T > factory, int direction
 		return targetType;
 	}
 
-	@Override public void process( RandomAccessible< ? extends T > source, RandomAccessibleInterval< ? extends T > target )
+	@Override
+	protected void process( RandomAccessible< ? extends T > source, RandomAccessibleInterval< ? extends T > target, ExecutorService executorService, int numThreads )
 	{
-		internConvolve( factory, Views.interval( source, requiredSourceInterval( target ) ), target, direction );
-	}
-
-	private < T > void internConvolve( LineConvolverFactory< ? super T > factory, RandomAccessibleInterval< ? extends T > source, RandomAccessibleInterval< ? extends T > target, int d )
-	{
-		final long[] sourceMin = Intervals.minAsLongArray( source );
+		RandomAccessibleInterval< ? extends T > sourceInterval = Views.interval( source, requiredSourceInterval( target ) );
+		final long[] sourceMin = Intervals.minAsLongArray( sourceInterval );
 		final long[] targetMin = Intervals.minAsLongArray( target );
 
 		Supplier< Consumer< Localizable > > actionFactory = () -> {
 
-			final RandomAccess< ? extends T > in = source.randomAccess();
+			final RandomAccess< ? extends T > in = sourceInterval.randomAccess();
 			final RandomAccess< ? extends T > out = target.randomAccess();
-			final Runnable convolver = factory.getConvolver( in, out, d, target.dimension( d ) );
+			final Runnable convolver = factory.getConvolver( in, out, direction, target.dimension( direction ) );
 
 			return position -> {
 				in.setPosition( sourceMin );
@@ -79,12 +76,9 @@ private < T > void internConvolve( LineConvolverFactory< ? super T > factory, Ra
 		};
 
 		final long[] dim = Intervals.dimensionsAsLongArray( target );
-		dim[ d ] = 1;
+		dim[ direction ] = 1;
 
-		// FIXME: is there a better way to determine the number of threads
-		final int numThreads = getNumThreads();
 		final int numTasks = numThreads > 1 ? numThreads * 4 : 1;
-		ExecutorService executorService = getExecutor();
 		LineConvolution.forEachIntervalElementInParallel( executorService, numTasks, new FinalInterval( dim ), actionFactory );
 	}
 

src/test/java/net/imglib2/algorithm/convolution/GaussBenchmark.java

@@ -8,6 +8,8 @@
 import net.imglib2.algorithm.gauss3.SeparableSymmetricConvolution;
 import net.imglib2.img.Img;
 import net.imglib2.img.array.ArrayImgs;
+import net.imglib2.type.numeric.NumericType;
+import net.imglib2.type.numeric.RealType;
 import net.imglib2.type.numeric.real.DoubleType;
 import net.imglib2.view.Views;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -54,14 +56,14 @@ public void benchmarkSeparableSymmertricConvolution()
 	@Benchmark
 	public void benchmarkFastGauss()
 	{
-		FastGauss.convolve( sigma, inImage, outImage );
+		FastGauss.convolution( sigma ).process( inImage, outImage );
 	}
 
 	public static void main( String[] args ) throws RunnerException
 	{
 		Options opt = new OptionsBuilder()
 				.include( GaussBenchmark.class.getSimpleName() )
-				.forks( 0 )
+				.forks( 1 )
 				.warmupIterations( 4 )
 				.measurementIterations( 8 )
 				.warmupTime( TimeValue.milliseconds( 100 ) )