Use LoopBuilder and Parallelization classes in Gauss3

This simplifies the code, but also improves the performance,
because Gauss3.gauss(sigma, source, target) will no longer create
it's own ExecutorService.

This also means that the multi threading behaviour of Gauss3
can now conviniently be controled using the Parallelization class.
For example:

Parallelization.runSingleThreaded(() -> {
	Gauss3.gauss(sigma, source, target);
});

Author: maarzt

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

@@ -51,6 +51,7 @@
  *
  * @author Matthias Arzt
  */
+@Deprecated
 public abstract class AbstractMultiThreadedConvolution< T > implements Convolution< T >
 {
 
@@ -61,6 +62,7 @@ abstract protected void process( RandomAccessible< ? extends T > source,
 			ExecutorService executorService,
 			int numThreads );
 
+	@Deprecated
 	@Override
 	public void setExecutor( final ExecutorService executor )
 	{

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

@@ -64,10 +64,12 @@ class Concatenation< T > implements Convolution< T >
 		this.steps = new ArrayList<>( steps );
 	}
 
+	@Deprecated
 	@Override
-	public void setExecutor( final ExecutorService executor )
+	public void setExecutor( ExecutorService executor )
 	{
-		steps.forEach( step -> step.setExecutor( executor ) );
+		for ( Convolution<T> step : steps )
+			step.setExecutor( executor );
 	}
 
 	@Override

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

@@ -67,6 +67,7 @@ public interface Convolution< T >
 	/**
 	 * Set the {@link ExecutorService} to be used for convolution.
 	 */
+	@Deprecated
 	default void setExecutor( final ExecutorService executor )
 	{}
 

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

@@ -33,44 +33,50 @@
  */
 package net.imglib2.algorithm.convolution;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.function.Consumer;
-import java.util.function.Supplier;
-
 import net.imglib2.FinalInterval;
 import net.imglib2.Interval;
 import net.imglib2.Localizable;
-import net.imglib2.Point;
 import net.imglib2.RandomAccess;
 import net.imglib2.RandomAccessible;
 import net.imglib2.RandomAccessibleInterval;
-import net.imglib2.util.IntervalIndexer;
+import net.imglib2.loops.LoopBuilder;
+import net.imglib2.parallel.Parallelization;
+import net.imglib2.parallel.TaskExecutor;
+import net.imglib2.parallel.TaskExecutors;
+import net.imglib2.util.Cast;
 import net.imglib2.util.Intervals;
+import net.imglib2.util.Localizables;
 import net.imglib2.view.Views;
 
+import java.util.concurrent.ExecutorService;
+
 /**
  * This class can be used to implement a separable convolution. It applies a
  * {@link LineConvolverFactory} on the given images.
  *
  * @author Matthias Arzt
  */
-public class LineConvolution< T > extends AbstractMultiThreadedConvolution< T >
+public class LineConvolution< T > implements Convolution<T>
 {
 	private final LineConvolverFactory< ? super T > factory;
 
 	private final int direction;
 
+	private ExecutorService executor;
+
 	public LineConvolution( final LineConvolverFactory< ? super T > factory, final int direction )
 	{
 		this.factory = factory;
 		this.direction = direction;
 	}
 
+	@Deprecated
+	@Override
+	public void setExecutor( ExecutorService executor )
+	{
+		this.executor = executor;
+	}
+
 	@Override
 	public Interval requiredSourceInterval( final Interval targetInterval )
 	{
@@ -84,104 +90,38 @@ public Interval requiredSourceInterval( final Interval targetInterval )
 	@Override
 	public T preferredSourceType( final T targetType )
 	{
-		return (T) factory.preferredSourceType( targetType );
+		return Cast.unchecked( factory.preferredSourceType( targetType ) );
 	}
 
 	@Override
-	protected void process( final RandomAccessible< ? extends T > source, final RandomAccessibleInterval< ? extends T > target, final ExecutorService executorService, final int numThreads )
+	public void process( RandomAccessible< ? extends T > source, RandomAccessibleInterval< ? extends T > target )
 	{
 		final RandomAccessibleInterval< ? extends T > sourceInterval = Views.interval( source, requiredSourceInterval( target ) );
 		final long[] sourceMin = Intervals.minAsLongArray( sourceInterval );
 		final long[] targetMin = Intervals.minAsLongArray( target );
 
-		final Supplier< Consumer< Localizable > > actionFactory = () -> {
-
-			final RandomAccess< ? extends T > in = sourceInterval.randomAccess();
-			final RandomAccess< ? extends T > out = target.randomAccess();
-			final Runnable convolver = factory.getConvolver( in, out, direction, target.dimension( direction ) );
-
-			return position -> {
-				in.setPosition( sourceMin );
-				out.setPosition( targetMin );
-				in.move( position );
-				out.move( position );
-				convolver.run();
-			};
-		};
-
 		final long[] dim = Intervals.dimensionsAsLongArray( target );
 		dim[ direction ] = 1;
 
-		final int numTasks = numThreads > 1 ? timesFourAvoidOverflow(numThreads) : 1;
-		LineConvolution.forEachIntervalElementInParallel( executorService, numTasks, new FinalInterval( dim ), actionFactory );
-	}
+		RandomAccessibleInterval< Localizable > positions = Localizables.randomAccessibleInterval( new FinalInterval( dim ) );
+		TaskExecutor taskExecutor = executor == null ? Parallelization.getTaskExecutor() : TaskExecutors.forExecutorService( executor );
+		LoopBuilder.setImages( positions ).multiThreaded(taskExecutor).forEachChunk(
+				chunk -> {
 
-	private int timesFourAvoidOverflow( int x )
-	{
-		return (int) Math.min((long) x * 4, Integer.MAX_VALUE);
-	}
+					final RandomAccess< ? extends T > in = sourceInterval.randomAccess();
+					final RandomAccess< ? extends T > out = target.randomAccess();
+					final Runnable convolver = factory.getConvolver( in, out, direction, target.dimension( direction ) );
 
-	/**
-	 * {@link #forEachIntervalElementInParallel(ExecutorService, int, Interval, Supplier)}
-	 * executes a given action for each position in a given interval. Therefor
-	 * it starts the specified number of tasks. Each tasks calls the action
-	 * factory once, to get an instance of the action that should be executed.
-	 * The action is then called multiple times by the task.
-	 *
-	 * @param service
-	 *            {@link ExecutorService} used to create the tasks.
-	 * @param numTasks
-	 *            number of tasks to use.
-	 * @param interval
-	 *            interval to iterate over.
-	 * @param actionFactory
-	 *            factory that returns the action to be executed.
-	 */
-	// TODO: move to a better place
-	public static void forEachIntervalElementInParallel( final ExecutorService service, final int numTasks, final Interval interval,
-			final Supplier< Consumer< Localizable > > actionFactory )
-	{
-		final long[] min = Intervals.minAsLongArray( interval );
-		final long[] dim = Intervals.dimensionsAsLongArray( interval );
-		final long size = Intervals.numElements( dim );
-		final int boundedNumTasks = (int) Math.max( 1, Math.min(size, numTasks ));
-		final long taskSize = ( size - 1 ) / boundedNumTasks + 1; // taskSize = roundUp(size / boundedNumTasks);
-		final ArrayList< Callable< Void > > callables = new ArrayList<>();
+					chunk.forEachPixel( position -> {
+						in.setPosition( sourceMin );
+						out.setPosition( targetMin );
+						in.move( position );
+						out.move( position );
+						convolver.run();
+					} );
 
-		for ( int taskNum = 0; taskNum < boundedNumTasks; ++taskNum )
-		{
-			final long myStartIndex = taskNum * taskSize;
-			final long myEndIndex = Math.min( size, myStartIndex + taskSize );
-			final Callable< Void > r = () -> {
-				final Consumer< Localizable > action = actionFactory.get();
-				final long[] position = new long[ dim.length ];
-				final Localizable localizable = Point.wrap( position );
-				for ( long index = myStartIndex; index < myEndIndex; ++index )
-				{
-					IntervalIndexer.indexToPositionWithOffset( index, dim, min, position );
-					action.accept( localizable );
+					return null;
 				}
-				return null;
-			};
-			callables.add( r );
-		}
-		execute( service, callables );
-	}
-
-	private static void execute( final ExecutorService service, final ArrayList< Callable< Void > > callables )
-	{
-		try
-		{
-			final List< Future< Void > > futures = service.invokeAll( callables );
-			for ( final Future< Void > future : futures )
-				future.get();
-		}
-		catch ( final InterruptedException | ExecutionException e )
-		{
-			final Throwable cause = e.getCause();
-			if ( cause instanceof RuntimeException )
-				throw ( RuntimeException ) cause;
-			throw new RuntimeException( e );
-		}
+		);
 	}
 }

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

@@ -53,6 +53,7 @@ public class MultiDimensionConvolution< T > implements Convolution< T >
 {
 	private ExecutorService executor;
 
+	@Deprecated
 	@Override
 	public void setExecutor( final ExecutorService executor )
 	{

src/main/java/net/imglib2/algorithm/gauss3/Gauss3.java

@@ -43,6 +43,7 @@
 import net.imglib2.algorithm.convolution.kernel.Kernel1D;
 import net.imglib2.algorithm.convolution.kernel.SeparableKernelConvolution;
 import net.imglib2.exception.IncompatibleTypeException;
+import net.imglib2.parallel.Parallelization;
 import net.imglib2.type.numeric.NumericType;
 import net.imglib2.type.numeric.RealType;
 import net.imglib2.type.numeric.real.DoubleType;
@@ -126,10 +127,9 @@ public static < S extends NumericType< S >, T extends NumericType< T > > void ga
 	 */
 	public static < S extends NumericType< S >, T extends NumericType< T > > void gauss( final double[] sigma, final RandomAccessible< S > source, final RandomAccessibleInterval< T > target ) throws IncompatibleTypeException
 	{
-		final int numthreads = Runtime.getRuntime().availableProcessors();
-		final ExecutorService service = Executors.newFixedThreadPool( numthreads );
-		gauss( sigma, source, target, service );
-		service.shutdown();
+		final double[][] halfkernels = halfkernels( sigma );
+		final Convolution< NumericType< ? > > convolution = SeparableKernelConvolution.convolution( Kernel1D.symmetric( halfkernels ) );
+		convolution.process( source, target );
 	}
 
 	/**
@@ -201,10 +201,9 @@ public static < S extends NumericType< S >, T extends NumericType< T > > void ga
 	 */
 	public static < S extends NumericType< S >, T extends NumericType< T > > void gauss( final double[] sigma, final RandomAccessible< S > source, final RandomAccessibleInterval< T > target, final ExecutorService service ) throws IncompatibleTypeException
 	{
-		final double[][] halfkernels = halfkernels( sigma );
-		final Convolution< NumericType< ? > > convolution = SeparableKernelConvolution.convolution( Kernel1D.symmetric( halfkernels ) );
-		convolution.setExecutor( service );
-		convolution.process( source, target );
+		Parallelization.runWithExecutor( service,
+				() -> gauss( sigma, source, target )
+		);
 	}
 
 	public static double[][] halfkernels( final double[] sigma )

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

@@ -37,11 +37,8 @@
 import net.imglib2.Interval;
 import net.imglib2.RandomAccessible;
 import net.imglib2.RandomAccessibleInterval;
-import net.imglib2.algorithm.convolution.kernel.SeparableKernelConvolution;
 import net.imglib2.img.Img;
-import net.imglib2.img.ImgFactory;
 import net.imglib2.img.array.ArrayImgs;
-import net.imglib2.img.cell.CellImgFactory;
 import net.imglib2.loops.LoopBuilder;
 import net.imglib2.type.numeric.RealType;
 import net.imglib2.type.numeric.integer.IntType;
@@ -51,6 +48,7 @@
 import net.imglib2.util.Intervals;
 import net.imglib2.view.IntervalView;
 import net.imglib2.view.Views;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import static org.junit.Assert.assertArrayEquals;
@@ -88,6 +86,7 @@ public void testDifferences2()
 		assertArrayEquals( new int[] { -3, 3 }, targetPixels );
 	}
 
+	@Ignore( "takes to long" )
 	@Test
 	public void testHugeImage()
 	{
@@ -96,7 +95,7 @@ public void testHugeImage()
 		assertTrue( width * height > Integer.MAX_VALUE );
 		RandomAccessible< UnsignedByteType > source = ConstantUtils.constantRandomAccessible( new UnsignedByteType(), 2 );
 		RandomAccessibleInterval< UnsignedByteType > target = ConstantUtils.constantRandomAccessibleInterval(
-				new UnsignedByteType(), 2, new FinalInterval( width, height ) );
+				new UnsignedByteType(), new FinalInterval( width, height ) );
 		double[][] kernels = { { 2 }, { 3 } };
 		try
 		{

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

@@ -37,6 +37,8 @@
 import net.imglib2.RandomAccess;
 import net.imglib2.img.Img;
 import net.imglib2.img.array.ArrayImgs;
+import net.imglib2.parallel.Parallelization;
+import net.imglib2.parallel.TaskExecutors;
 import net.imglib2.type.numeric.integer.UnsignedByteType;
 import net.imglib2.util.Intervals;
 import org.junit.Test;
@@ -85,8 +87,11 @@ public void testNumTasksEqualsIntegerMaxValue() {
 		byte[] result = new byte[ 1 ];
 		Img< UnsignedByteType > out = ArrayImgs.unsignedBytes( result, result.length );
 		Img< UnsignedByteType > in = ArrayImgs.unsignedBytes( new byte[] { 1, 2 }, 2 );
-		final LineConvolution< UnsignedByteType > convolution = new LineConvolution<>( new ForwardDifferenceConvolverFactory(), 0 );
-		convolution.process( in, out, Executors.newSingleThreadExecutor(), Integer.MAX_VALUE );
+		Runnable runnable = () -> {
+			final LineConvolution< UnsignedByteType > convolution = new LineConvolution<>( new ForwardDifferenceConvolverFactory(), 0 );
+			convolution.process( in, out );
+		};
+		Parallelization.runWithExecutor( TaskExecutors.forExecutorServiceAndNumTasks( Executors.newSingleThreadExecutor(), Integer.MAX_VALUE) , runnable );
 		assertArrayEquals( new byte[] { 1 }, result );
 	}