Scope failures locally

Author: janniklinde

src/main/java/org/apache/sysds/api/DMLScript.java

@@ -444,9 +444,6 @@ private static void execute(String dmlScriptStr, String fnameOptConfig, Map<Stri
 
 		// optionally register for monitoring
 		registerForMonitoring();
-
-		// reset any errors from the LocalTaskQueue
-		LocalTaskQueue.resetFailures();
 
 		//Step 1: parse configuration files & write any configuration specific global variables
 		loadConfiguration(fnameOptConfig);

src/main/java/org/apache/sysds/runtime/controlprogram/parfor/LocalTaskQueue.java

@@ -42,10 +42,10 @@ public class LocalTaskQueue<T>
 
 	public static final int    MAX_SIZE      = 100000; //main memory constraint
 	public static final Object NO_MORE_TASKS = null; //object to signal NO_MORE_TASKS
-	private static volatile DMLRuntimeException FAILURE = null;
 
 	private LinkedList<T>  _data        = null;
 	private boolean 	   _closedInput = false;
+	private DMLRuntimeException _failure = null;
 	private static final Log LOG = LogFactory.getLog(LocalTaskQueue.class.getName());
 
 	public LocalTaskQueue()
@@ -63,14 +63,14 @@ public LocalTaskQueue()
 	public synchronized void enqueueTask( T t ) 
 		throws InterruptedException
 	{
-		while( _data.size() + 1 > MAX_SIZE && FAILURE == null )
+		while( _data.size() + 1 > MAX_SIZE && _failure == null )
 		{
 			LOG.warn("MAX_SIZE of task queue reached.");
 			wait(); //max constraint reached, wait for read
 		}
 
-		if ( FAILURE != null )
-			throw FAILURE;
+		if ( _failure != null )
+			throw _failure;
 
 		_data.addLast( t );
 
@@ -87,16 +87,16 @@ public synchronized void enqueueTask( T t )
 	public synchronized T dequeueTask() 
 		throws InterruptedException
 	{
-		while( _data.isEmpty() && FAILURE == null )
+		while( _data.isEmpty() && _failure == null )
 		{
 			if( !_closedInput )
 				wait(); // wait for writers
 			else
 				return (T)NO_MORE_TASKS; 
 		}
 
-		if ( FAILURE != null )
-			throw FAILURE;
+		if ( _failure != null )
+			throw _failure;
 
 		T t = _data.removeFirst();
 
@@ -119,8 +119,11 @@ public synchronized boolean isProcessed() {
 		return _closedInput && _data.isEmpty();
 	}
 
-	public synchronized void notifyFailure() {
-		notifyAll();
+	public synchronized void propagateFailure(DMLRuntimeException failure) {
+		if (_failure == null) {
+			_failure = failure;
+			notifyAll();
+		}
 	}
 
 	@Override
@@ -147,18 +150,4 @@ public synchronized String toString()
 
 		return sb.toString();
 	}
-
-	public static boolean failGlobally(DMLRuntimeException ex) {
-		// Only register the first failure
-		if (FAILURE == null) {
-			FAILURE = ex;
-			return true;
-		}
-
-		return false;
-	}
-
-	public static void resetFailures() {
-		FAILURE = null;
-	}
 }

src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java

@@ -111,10 +111,8 @@ private Runnable oocTask(Runnable r, LocalTaskQueue<?>... queues) {
 			catch (Exception ex) {
 				DMLRuntimeException re = ex instanceof DMLRuntimeException ? (DMLRuntimeException) ex : new DMLRuntimeException(ex);
 
-				LocalTaskQueue.failGlobally(re);
-
 				for (LocalTaskQueue<?> q : queues) {
-					q.notifyFailure();
+					q.propagateFailure(re);
 				}
 			}
 		};