[MINOR] Federated Timeout

This commit reduce the timeout for federated tests, and enforce the timeout on federated requests. Previously we had some test cases that would infinitely run, and therefore we would not be able to decipher the log messages (because nothing would be written). This commit change it by enforcing a strict 16 seconds execution time for a single federated requests and a 1 day timeout for a default federated requests. Previously some operations did use the federated timeout. However, it was not enforced in critical places.

Closes #2179

Author: Baunsgaard

src/main/java/org/apache/sysds/conf/DMLConfig.java

@@ -201,7 +201,7 @@ public class DMLConfig
 		_defaultVals.put(FLOATING_POINT_PRECISION, "double" );
 		_defaultVals.put(USE_SSL_FEDERATED_COMMUNICATION, "false");
 		_defaultVals.put(DEFAULT_FEDERATED_INITIALIZATION_TIMEOUT, "10");
-		_defaultVals.put(FEDERATED_TIMEOUT,      "-1");
+		_defaultVals.put(FEDERATED_TIMEOUT,      "86400"); // default 1 day compute timeout.
 		_defaultVals.put(FEDERATED_PLANNER,      FederatedPlanner.RUNTIME.name());
 		_defaultVals.put(FEDERATED_PAR_CONN,     "-1"); // vcores
 		_defaultVals.put(FEDERATED_PAR_INST,     "-1"); // vcores

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedStatistics.java

@@ -503,7 +503,8 @@ public static List<DataObjectModel> getWorkerDataObjects() {
 		return new ArrayList<>(workerDataObjects.values());
 	}
 
-	public static void addEvent(EventModel event) {
+	public synchronized static void addEvent(EventModel event) { 
+		// synchronized, because multiple requests can be handled concurrently
 		workerEvents.add(event);
 	}
 

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java

@@ -651,8 +651,7 @@ private FederatedResponse execUDF(FederatedRequest request, ExecutionContextMap
 		// get function and input parameters
 		try {
 			FederatedUDF udf = (FederatedUDF) request.getParam(0);
-			if(LOG.isDebugEnabled())
-				LOG.debug(udf);
+			LOG.debug(udf);
 
 			eventStage.operation = udf.getClass().getSimpleName();
 

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java

@@ -26,14 +26,17 @@
 import java.util.Map.Entry;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.conf.ConfigurationManager;
 import org.apache.sysds.hops.fedplanner.FTypes.AlignType;
 import org.apache.sysds.hops.fedplanner.FTypes.FType;
 import org.apache.sysds.lops.RightIndex;
@@ -637,11 +640,25 @@ public long getMaxIndexInRange(int dim) {
 	 * @param forEachFunction function to execute for each pair
 	 */
 	public void forEachParallel(BiFunction<FederatedRange, FederatedData, Void> forEachFunction) {
-		ExecutorService pool = CommonThreadPool.get(_fedMap.size());
+		ExecutorService pool = Executors.newFixedThreadPool(_fedMap.size());
 		ArrayList<MappingTask> mappingTasks = new ArrayList<>();
 		for(Pair<FederatedRange, FederatedData> fedMap : _fedMap)
 			mappingTasks.add(new MappingTask(fedMap.getKey(), fedMap.getValue(), forEachFunction, _ID));
-		CommonThreadPool.invokeAndShutdown(pool, mappingTasks);
+
+		try {
+			for(Future<?> t:pool.invokeAll(mappingTasks, ConfigurationManager.getFederatedTimeout(), TimeUnit.SECONDS)){
+				if(!t.isDone())
+					throw new RuntimeException("Timeout");
+				else if(t.isCancelled())
+					throw new RuntimeException("Failed");
+			}
+		}
+		catch(InterruptedException e) {
+			throw new RuntimeException(e);
+		}
+		finally{
+			pool.shutdown();
+		}
 	}
 
 	/**
@@ -655,15 +672,25 @@ public void forEachParallel(BiFunction<FederatedRange, FederatedData, Void> forE
 	 * @return the new <code>FederationMap</code>
 	 */
 	public FederationMap mapParallel(long newVarID, BiFunction<FederatedRange, FederatedData, Void> mappingFunction) {
-		ExecutorService pool = CommonThreadPool.get(_fedMap.size());
-
+		ExecutorService pool = Executors.newFixedThreadPool(_fedMap.size());
 		FederationMap fedMapCopy = copyWithNewID(_ID);
 		ArrayList<MappingTask> mappingTasks = new ArrayList<>();
 		for(Pair<FederatedRange, FederatedData> fedMap : fedMapCopy._fedMap)
 			mappingTasks.add(new MappingTask(fedMap.getKey(), fedMap.getValue(), mappingFunction, newVarID));
-		CommonThreadPool.invokeAndShutdown(pool, mappingTasks);
-		fedMapCopy._ID = newVarID;
-		return fedMapCopy;
+		try{
+			for(Future<?> t : pool.invokeAll(mappingTasks, ConfigurationManager.getFederatedTimeout(), TimeUnit.SECONDS)){
+				if(!t.isDone())
+					throw new RuntimeException("Timeout");
+				else if(t.isCancelled()){
+					throw new RuntimeException("Failed");
+				}
+			}
+			fedMapCopy._ID = newVarID;
+			return fedMapCopy;
+		}
+		catch(Exception e){
+			throw new RuntimeException(e);
+		}
 	}
 
 	public FederationMap filter(IndexRange ixrange) {

src/main/java/org/apache/sysds/runtime/instructions/cp/ParamservBuiltinCPInstruction.java

@@ -51,6 +51,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -61,6 +62,7 @@
 import org.apache.spark.util.LongAccumulator;
 import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.conf.ConfigurationManager;
 import org.apache.sysds.hops.recompile.Recompiler;
 import org.apache.sysds.parser.Statement.FederatedPSScheme;
 import org.apache.sysds.parser.Statement.PSFrequency;
@@ -241,13 +243,16 @@ model, aggServiceEC, getValFunction(), getNumBatchesPerEpoch(runtimeBalancing, r
 
 		try {
 			// Launch the worker threads and wait for completion
-			for (Future<Void> ret : es.invokeAll(threads))
-				ret.get(); //error handling
+			for (Future<Void> ret : es.invokeAll(threads, ConfigurationManager.getFederatedTimeout(), TimeUnit.SECONDS)){
+				if(!ret.isDone())
+					throw new RuntimeException("Failed federated execution");
+				// ret.get(); //error handling
+			}
 			// Fetch the final model from ps
 			ec.setVariable(output.getName(), ps.getResult());
 			if (DMLScript.STATISTICS)
 				ParamServStatistics.accExecutionTime((long) ParamServStatistics.getExecutionTimer().stop());
-		} catch (InterruptedException | ExecutionException e) {
+		} catch (Exception e) {
 			throw new DMLRuntimeException("ParamservBuiltinCPInstruction: unknown error: ", e);
 		} finally {
 			es.shutdownNow();

src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java

@@ -35,6 +35,7 @@
 import org.apache.sysds.common.Types;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.hops.OptimizerUtils;
 import org.apache.sysds.hops.fedplanner.FTypes;
 import org.apache.sysds.hops.fedplanner.FTypes.FType;
 import org.apache.sysds.lops.PickByCount;
@@ -47,10 +48,10 @@
 import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.RequestType;
 import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse;
 import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse.ResponseType;
-import org.apache.sysds.runtime.frame.data.FrameBlock;
 import org.apache.sysds.runtime.controlprogram.federated.FederatedUDF;
 import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
 import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
+import org.apache.sysds.runtime.frame.data.FrameBlock;
 import org.apache.sysds.runtime.instructions.InstructionUtils;
 import org.apache.sysds.runtime.instructions.cp.CPOperand;
 import org.apache.sysds.runtime.instructions.cp.Data;
@@ -175,6 +176,7 @@ public void processInstruction(ExecutionContext ec) {
 				try {
 					FederatedResponse response = responseFuture.get();
 					MultiColumnEncoder encoder = (MultiColumnEncoder) response.getData()[0];
+
 					// merge this encoder into a composite encoder
 					synchronized(finalGlobalEncoder) {
 						finalGlobalEncoder.mergeAt(encoder, columnOffset, (int) (range.getBeginDims()[0] + 1));
@@ -378,24 +380,30 @@ public ExecuteFrameEncoder(long input, long output, MultiColumnEncoder encoder)
 
 		@Override
 		public FederatedResponse execute(ExecutionContext ec, Data... data) {
-			FrameBlock fb = ((FrameObject) data[0]).acquireReadAndRelease();
-
-			// offset is applied on the Worker to shift the local encoders to their respective column
-			_encoder.applyColumnOffset();
-			// apply transformation
-			//MatrixBlock mbout = _encoder.apply(fb, OptimizerUtils.getTransformNumThreads());
-			// FIXME: Enabling multithreading intermittently hangs
-			MatrixBlock mbout = _encoder.apply(fb, 1);
-
-			// create output matrix object
-			MatrixObject mo = ExecutionContext.createMatrixObject(mbout);
-
-			// add it to the list of variables
-			ec.setVariable(String.valueOf(_outputID), mo);
+			try{
 
-			// return id handle
-			return new FederatedResponse(
-				ResponseType.SUCCESS_EMPTY, mbout.getNonZeros());
+				FrameBlock fb = ((FrameObject) data[0]).acquireReadAndRelease();
+	
+				// offset is applied on the Worker to shift the local encoders to their respective column
+				_encoder.applyColumnOffset();
+				// apply transformation
+				MatrixBlock mbout = _encoder.apply(fb, OptimizerUtils.getTransformNumThreads());
+				// FIXME: Enabling multithreading intermittently hangs
+				// MatrixBlock mbout = _encoder.apply(fb, 1);
+	
+				// create output matrix object
+				MatrixObject mo = ExecutionContext.createMatrixObject(mbout);
+	
+				// add it to the list of variables
+				ec.setVariable(String.valueOf(_outputID), mo);
+	
+				// return id handle
+				return new FederatedResponse(
+					ResponseType.SUCCESS_EMPTY, mbout.getNonZeros());
+			}
+			catch(Exception e){
+				return new FederatedResponse(ResponseType.ERROR);
+			}
 		}
 
 		@Override

src/test/config/SystemDS-MultiTenant-config.xml

@@ -21,6 +21,6 @@
    <!-- The timeout of the federated tests to initialize the federated matrixes -->
    <sysds.federated.initialization.timeout>30</sysds.federated.initialization.timeout>
    <!-- The timeout of each instruction sent to federated workers -->
-   <sysds.federated.timeout>128</sysds.federated.timeout>
+   <sysds.federated.timeout>16</sysds.federated.timeout>
    <sysds.local.spark>true</sysds.local.spark>
 </root>

src/test/config/SystemDS-config.xml

@@ -23,5 +23,5 @@
    <!-- The timeout of the federated tests to initialize the federated matrixes -->
    <sysds.federated.initialization.timeout>2</sysds.federated.initialization.timeout>
    <!-- The timeout of each instruction sent to federated workers -->
-   <sysds.federated.timeout>128</sysds.federated.timeout>
+   <sysds.federated.timeout>16</sysds.federated.timeout>
 </root>

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
 		{

src/test/java/org/apache/sysds/test/functions/federated/primitives/part3/FederatedTokenizeTest.java

@@ -79,6 +79,7 @@ public void testTokenizeFullDenseFrameCP() {
 
 	private void runAggregateOperationTest(ExecMode execMode) {
 		setExecMode(execMode);
+		setOutputBuffering(true);
 
 		String TEST_NAME = TEST_NAME1;