Merge pull request #78 from imglib/fix-issue-77

Fix issue 77

Author: tpietzsch

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

@@ -39,7 +39,7 @@ final public void process( final RandomAccessible< ? extends T > source, final R
 	{
 		if ( executor == null )
 		{
-			final int numThreads = suggestNumThreads();
+			final int numThreads = Runtime.getRuntime().availableProcessors();
 			final ExecutorService executor = Executors.newFixedThreadPool( numThreads );
 			try
 			{
@@ -56,15 +56,13 @@ final public void process( final RandomAccessible< ? extends T > source, final R
 		}
 	}
 
-	private int getNumThreads( final ExecutorService executor )
+	static int getNumThreads( final ExecutorService executor )
 	{
-		if ( executor instanceof ThreadPoolExecutor )
-			return ( ( ThreadPoolExecutor ) executor ).getMaximumPoolSize();
-		return suggestNumThreads();
+		int maxPoolSize = ( executor instanceof ThreadPoolExecutor ) ?
+				( ( ThreadPoolExecutor ) executor ).getMaximumPoolSize() :
+				Integer.MAX_VALUE;
+		int availableProcessors = Runtime.getRuntime().availableProcessors();
+		return Math.max(1, Math.min(availableProcessors, maxPoolSize));
 	}
 
-	private int suggestNumThreads()
-	{
-		return Runtime.getRuntime().availableProcessors();
-	}
 }

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

@@ -16,10 +16,6 @@
 import net.imglib2.RandomAccess;
 import net.imglib2.RandomAccessible;
 import net.imglib2.RandomAccessibleInterval;
-import net.imglib2.type.Type;
-import net.imglib2.type.numeric.RealType;
-import net.imglib2.type.numeric.real.DoubleType;
-import net.imglib2.type.numeric.real.FloatType;
 import net.imglib2.util.IntervalIndexer;
 import net.imglib2.util.Intervals;
 import net.imglib2.view.Views;
@@ -83,10 +79,15 @@ protected void process( final RandomAccessible< ? extends T > source, final Rand
 		final long[] dim = Intervals.dimensionsAsLongArray( target );
 		dim[ direction ] = 1;
 
-		final int numTasks = numThreads > 1 ? numThreads * 4 : 1;
+		final int numTasks = numThreads > 1 ? timesFourAvoidOverflow(numThreads) : 1;
 		LineConvolution.forEachIntervalElementInParallel( executorService, numTasks, new FinalInterval( dim ), actionFactory );
 	}
 
+	private int timesFourAvoidOverflow( int x )
+	{
+		return (int) Math.min((long) x * 4, Integer.MAX_VALUE);
+	}
+
 	/**
 	 * {@link #forEachIntervalElementInParallel(ExecutorService, int, Interval, Supplier)}
 	 * executes a given action for each position in a given interval. Therefor
@@ -110,14 +111,14 @@ public static void forEachIntervalElementInParallel( final ExecutorService servi
 		final long[] min = Intervals.minAsLongArray( interval );
 		final long[] dim = Intervals.dimensionsAsLongArray( interval );
 		final long size = Intervals.numElements( dim );
-		final long endIndex = size;
-		final long taskSize = ( size + numTasks - 1 ) / numTasks; // round up
+		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<>();
 
-		for ( int taskNum = 0; taskNum < numTasks; ++taskNum )
+		for ( int taskNum = 0; taskNum < boundedNumTasks; ++taskNum )
 		{
 			final long myStartIndex = taskNum * taskSize;
-			final long myEndIndex = Math.min( endIndex, myStartIndex + 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 ];

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

@@ -0,0 +1,35 @@
+package net.imglib2.algorithm.convolution;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.junit.Test;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+/**
+ * Tests {@link AbstractMultiThreadedConvolution}
+ *
+ * @author Matthias Arzt
+ */
+public class AbstractMultiThreadedConvolutionTest
+{
+	@Test
+	public void testSuggestNumTasksFixedThreadPool() {
+		if ( Runtime.getRuntime().availableProcessors() < 3 )
+			return;
+		final ExecutorService executor = Executors.newFixedThreadPool( 3 );
+		int result = AbstractMultiThreadedConvolution.getNumThreads( executor );
+		assertEquals( 3, result );
+	}
+
+	@Test
+	public void testSuggestNumTasksCachedThreadPool() {
+		final ExecutorService executor = Executors.newCachedThreadPool();
+		int result = AbstractMultiThreadedConvolution.getNumThreads( executor );
+		assertTrue( Runtime.getRuntime().availableProcessors() >= result );
+	}
+}

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

@@ -8,6 +8,8 @@
 import net.imglib2.util.Intervals;
 import org.junit.Test;
 
+import java.util.concurrent.Executors;
+
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -45,6 +47,16 @@ public void testConvolve()
 		assertArrayEquals( expected, result );
 	}
 
+	@Test
+	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 );
+		assertArrayEquals( new byte[] { 1 }, result );
+	}
+
 	static class ForwardDifferenceConvolverFactory implements LineConvolverFactory< UnsignedByteType >
 	{