Update ConntectedComponenets to make use of the Parallelization context.

No new methods are added. Existing methods behave the same, but can be
configured using the Parallelization context.

Author: maarzt

src/main/java/net/imglib2/algorithm/labeling/ConnectedComponents.java

@@ -39,19 +39,16 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 
 import net.imglib2.Cursor;
 import net.imglib2.FinalInterval;
 import net.imglib2.RandomAccess;
 import net.imglib2.RandomAccessible;
 import net.imglib2.RandomAccessibleInterval;
 import net.imglib2.iterator.IntervalIterator;
+import net.imglib2.parallel.Parallelization;
 import net.imglib2.roi.labeling.ImgLabeling;
 import net.imglib2.type.numeric.IntegerType;
 import net.imglib2.view.Views;
@@ -99,17 +96,19 @@ public CollectNeighborLabelsFactory getFactory()
 	 * @param se
 	 *            structuring element to use. 8-connected or 4-connected
 	 *            (respectively n-dimensional analog)
+	 * @param service
+	 *            service providing threads for multi-threading
 	 */
 	public static < T extends IntegerType< T >, L, I extends IntegerType< I > > void labelAllConnectedComponents(
 			final RandomAccessible< T > input,
 			final ImgLabeling< L, I > labeling,
 			final Iterator< L > labelGenerator,
-			final StructuringElement se )
+			final StructuringElement se,
+			final ExecutorService service )
 	{
-		final int numThreads = Runtime.getRuntime().availableProcessors();
-		final ExecutorService service = Executors.newFixedThreadPool( numThreads );
-		labelAllConnectedComponents( input, labeling, labelGenerator, se, service );
-		service.shutdown();
+		Parallelization.runWithExecutor( service,
+				() -> labelAllConnectedComponents( input, labeling, labelGenerator, se )
+		);
 	}
 
 	/**
@@ -130,21 +129,18 @@ public static < T extends IntegerType< T >, L, I extends IntegerType< I > > void
 	 * @param se
 	 *            structuring element to use. 8-connected or 4-connected
 	 *            (respectively n-dimensional analog)
-	 * @param service
-	 *            service providing threads for multi-threading
 	 */
 	public static < T extends IntegerType< T >, L, I extends IntegerType< I > > void labelAllConnectedComponents(
 			final RandomAccessible< T > input,
 			final ImgLabeling< L, I > labeling,
 			final Iterator< L > labelGenerator,
-			final StructuringElement se,
-			final ExecutorService service )
+			final StructuringElement se )
 	{
 		final RandomAccessibleInterval< I > output = labeling.getIndexImg();
 		for  ( final I i : Views.iterable( output ) )
 			i.setZero();
 
-		final int numLabels = labelAllConnectedComponents( input, output, se, service ) + 1;
+		final int numLabels = labelAllConnectedComponents( input, output, se ) + 1;
 
 		final ArrayList< Set< L > > labelSets = new ArrayList<>();
 		labelSets.add( Collections.emptySet() );
@@ -161,7 +157,6 @@ public static < T extends IntegerType< T >, L, I extends IntegerType< I > > void
 	 *
 	 * <p>
 	 * <em>Note, that the {@code output} image must be cleared to 0!</em>
-	 * </p>
 	 *
 	 * @param input
 	 *            input image with pixels &gt; 0 belonging to foreground.
@@ -170,19 +165,20 @@ public static < T extends IntegerType< T >, L, I extends IntegerType< I > > void
 	 * @param se
 	 *            structuring element to use. 8-connected or 4-connected
 	 *            (respectively n-dimensional analog)
+	 * @param service
+	 *            service providing threads for multi-threading
 	 * @return the number of connected components (that is, the highest value
 	 *         occurring in the output image.
 	 */
 	public static < T extends IntegerType< T >, L extends IntegerType< L > > int labelAllConnectedComponents(
 			final RandomAccessible< T > input,
 			final RandomAccessibleInterval< L > output,
-			final StructuringElement se )
+			final StructuringElement se,
+			final ExecutorService service )
 	{
-		final int numThreads = Runtime.getRuntime().availableProcessors();
-		final ExecutorService service = Executors.newFixedThreadPool( numThreads );
-		final int result = labelAllConnectedComponents( input, output, se, service );
-		service.shutdown();
-		return result;
+		return Parallelization.runWithExecutor( service,
+				() -> labelAllConnectedComponents( input, output, se )
+		);
 	}
 
 	/**
@@ -192,6 +188,7 @@ public static < T extends IntegerType< T >, L extends IntegerType< L > > int lab
 	 *
 	 * <p>
 	 * <em>Note, that the {@code output} image must be cleared to 0!</em>
+	 * </p>
 	 *
 	 * @param input
 	 *            input image with pixels &gt; 0 belonging to foreground.
@@ -200,16 +197,13 @@ public static < T extends IntegerType< T >, L extends IntegerType< L > > int lab
 	 * @param se
 	 *            structuring element to use. 8-connected or 4-connected
 	 *            (respectively n-dimensional analog)
-	 * @param service
-	 *            service providing threads for multi-threading
 	 * @return the number of connected components (that is, the highest value
 	 *         occurring in the output image.
 	 */
 	public static < T extends IntegerType< T >, L extends IntegerType< L > > int labelAllConnectedComponents(
 			final RandomAccessible< T > input,
 			final RandomAccessibleInterval< L > output,
-			final StructuringElement se,
-			final ExecutorService service )
+			final StructuringElement se )
 	{
 		final int n = output.numDimensions();
 		final int splitDim = n - 1;
@@ -234,37 +228,14 @@ public static < T extends IntegerType< T >, L extends IntegerType< L > > int lab
 			min[ splitDim ] += taskSize;
 		}
 
-		final ArrayList< Future< ? > > futures = new ArrayList< Future< ? > >();
-		for ( final Fragment< T, L > fragment : fragments )
-		{
-			futures.add( service.submit( new Runnable()
-			{
-				@Override
-				public void run()
-				{
-					fragment.mark();
-				}
-			} ) );
-		}
-		getAllFutures( futures );
+		Parallelization.getTaskExecutor().forEach( Arrays.asList( fragments ), Fragment::mark );
 
 		final TIntArrayList merged = mergeCanonicalLists( fragments );
 		for ( int i = 1; i < numTasks; ++i )
 			fragments[ i ].linkToPreviousFragment( fragments[ i - 1 ], merged );
 		final int numComponents = splitCanonicalLists( fragments, merged );
 
-		for ( final Fragment< T, L > fragment : fragments )
-		{
-			futures.add( service.submit( new Runnable()
-			{
-				@Override
-				public void run()
-				{
-					fragment.relabel();
-				}
-			} ) );
-		}
-		getAllFutures( futures );
+		Parallelization.getTaskExecutor().forEach( Arrays.asList( fragments ), Fragment::relabel );
 
 		return numComponents;
 	}
@@ -481,26 +452,6 @@ private static < T extends IntegerType< T >, L extends IntegerType< L > > int sp
 		return nextLabel - 1;
 	}
 
-	private static void getAllFutures( final List< Future< ? > > futures )
-	{
-		for ( final Future< ? > future : futures )
-		{
-			try
-			{
-				future.get();
-			}
-			catch ( final InterruptedException e )
-			{
-				e.printStackTrace();
-			}
-			catch ( final ExecutionException e )
-			{
-				e.printStackTrace();
-			}
-		}
-		futures.clear();
-	}
-
 	private static interface CollectNeighborLabels< L extends IntegerType< L > >
 	{
 		public void collect( RandomAccess< L > la, final TIntArrayList neighborLabels, final long[] labelsMin, final long[] labelsMax );