output buffer test

Author: Baunsgaard

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

@@ -644,12 +644,17 @@ public void forEachParallel(BiFunction<FederatedRange, FederatedData, Void> forE
 		ArrayList<MappingTask> mappingTasks = new ArrayList<>();
 		for(Pair<FederatedRange, FederatedData> fedMap : _fedMap)
 			mappingTasks.add(new MappingTask(fedMap.getKey(), fedMap.getValue(), forEachFunction, _ID));
+
 		try {
-			pool.invokeAll(mappingTasks, ConfigurationManager.getFederatedTimeout(), TimeUnit.SECONDS);
+			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) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
+			throw new RuntimeException(e);
 		}
 		finally{
 			pool.shutdown();
@@ -667,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/fed/MultiReturnParameterizedBuiltinFEDInstruction.java

@@ -25,7 +25,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Stream;
 import java.util.zip.Adler32;
@@ -36,8 +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.conf.ConfigurationManager;
-import org.apache.sysds.conf.DMLConfig;
+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;
@@ -50,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;
@@ -176,8 +174,9 @@ public void processInstruction(ExecutionContext ec) {
 					new CreateFrameEncoder(data.getVarID(), spec, columnOffset + 1)));
 				// collect responses with encoders
 				try {
-					FederatedResponse response = responseFuture.get(ConfigurationManager.getFederatedTimeout(), TimeUnit.SECONDS);
+					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));
@@ -381,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/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java

@@ -333,7 +333,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
 
 		for(Entry<Object, Integer> e : _rcdMap.entrySet()) {
 			out.writeUTF(e.getKey().toString());
-			out.writeLong(e.getValue());
+			out.writeInt(e.getValue());
 		}
 	}
 

src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java

@@ -364,10 +364,11 @@ public MatrixBlock apply(CacheBlock<?> in, MatrixBlock out, int outputCol, int k
 		// There should be a encoder for every column
 		if(hasLegacyEncoder() && !(in instanceof FrameBlock))
 			throw new DMLRuntimeException("LegacyEncoders do not support non FrameBlock Inputs");
-		int numEncoders = getFromAll(ColumnEncoderComposite.class, ColumnEncoder::getColID).size();
+		int numEncoders = getEncoders().size();
+		// getFromAll(ColumnEncoderComposite.class, ColumnEncoder::getColID).size();
 		if(in.getNumColumns() != numEncoders)
 			throw new DMLRuntimeException("Not every column in has a CompositeEncoder. Please make sure every column "
-				+ "has a encoder or slice the input accordingly");
+				+ "has a encoder or slice the input accordingly: num encoders:  " + getEncoders()  + " vs columns " + in.getNumColumns());
 		// TODO smart checks
 		// Block allocation for MT access
 		if(in.getNumRows() == 0)

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;