Single thread

Author: Baunsgaard

src/main/java/org/apache/sysds/runtime/util/CommonThreadPool.java

@@ -19,6 +19,7 @@
 
 package org.apache.sysds.runtime.util;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map.Entry;
@@ -29,6 +30,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -103,11 +105,15 @@ public static ExecutorService get() {
 	 * @return The executor with specified parallelism
 	 */
 	public synchronized static ExecutorService get(int k) {
+		if(k <= 1){
+			LOG.warn("Invalid to create thread pool with <= one thread returning single thread executor", new RuntimeException());
+			return new SameThreadExecutorService();
+		}
+
 		final Thread thisThread = Thread.currentThread();
 		final String threadName = thisThread.getName();
 		// Contains main, because we name our test threads TestRunner_main
 		final boolean mainThread = threadName.contains("main");
-
 		if(size == k && mainThread)
 			return shared; // use the default thread pool if main thread and max parallelism.
 		else if(mainThread || threadName.contains("PARFOR")) {
@@ -134,6 +140,8 @@ else if(mainThread || threadName.contains("PARFOR")) {
 				return new CommonThreadPool(new ForkJoinPool(k));
 			}
 			return Executors.newFixedThreadPool(k);
+
+
 		}
 
 	}
@@ -330,4 +338,110 @@ else if(name.contains("test"))
 		else
 			return false;
 	}
+
+
+	private static class SameThreadExecutorService implements ExecutorService {
+
+		@Override
+		public void execute(Runnable command) {
+			command.run();
+		}
+
+		@Override
+		public void shutdown() {
+			// nothing
+		}
+
+		@Override
+		public List<Runnable> shutdownNow() {
+			return new ArrayList<>();
+		}
+
+		@Override
+		public boolean isShutdown() {
+			return false;
+		}
+
+		@Override
+		public boolean isTerminated() {
+			return false;
+
+		}
+
+		@Override
+		public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+			return true;
+		}
+
+		@Override
+		public <T> Future<T> submit(Callable<T> task) {
+			return new FutureTask<>(task);
+		}
+
+		@Override
+		public <T> Future<T> submit(Runnable task, T result) {
+			return new FutureTask<>(() -> {
+				task.run();
+				return result;
+			});
+		}
+
+		@Override
+		public Future<?> submit(Runnable task) {
+			return new FutureTask<>(() -> {
+				task.run();
+				return null;
+			});
+		}
+
+		@Override
+		public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+			List<Future<T>> ret = new ArrayList<>();
+			for(Callable<T> t : tasks)
+				ret.add(new FutureTask<>(t));
+			return ret;
+		}
+
+		@Override
+		public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+			throws InterruptedException {
+			List<Future<T>> ret = new ArrayList<>();
+			for(Callable<T> t : tasks)
+				ret.add(new FutureTask<>(t));
+			return ret;
+		}
+
+		@Override
+		public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+			Exception e = null;
+			for(Callable<T> t : tasks) {
+				try {
+					T r = t.call();
+					return r;
+				}
+				catch(Exception ee) {
+					e = ee;
+				}
+
+			}
+			throw new ExecutionException("failed", e);
+		}
+
+		@Override
+		public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+			throws InterruptedException, ExecutionException, TimeoutException {
+			Exception e = null;
+			for(Callable<T> t : tasks) {
+				try {
+					T r = t.call();
+					return r;
+				}
+				catch(Exception ee) {
+					e = ee;
+				}
+
+			}
+			throw new ExecutionException("failed", e);
+		}
+	}
 }

src/test/java/org/apache/sysds/test/functions/codegen/RowVectorComparisonTest.java

@@ -128,6 +128,7 @@ private void testCodegenIntegration( String testname, boolean rewrites, ExecType
 	{	
 		boolean oldFlag = OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION;
 		ExecMode platformOld = setExecMode(instType);
+		setOutputBuffering(true);
 
 		try
 		{