diff --git a/src/main/java/org/apache/sysds/api/DMLScript.java b/src/main/java/org/apache/sysds/api/DMLScript.java
index d4743ed..635aef7 100644
--- a/src/main/java/org/apache/sysds/api/DMLScript.java
+++ b/src/main/java/org/apache/sysds/api/DMLScript.java
@@ -59,6 +59,7 @@
 import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
 import org.apache.sysds.runtime.controlprogram.context.ExecutionContextFactory;
 import org.apache.sysds.runtime.controlprogram.context.SparkExecutionContext;
+import org.apache.sysds.runtime.controlprogram.federated.FederatedData;
 import org.apache.sysds.runtime.controlprogram.federated.FederatedWorker;
 import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
 import org.apache.sysds.runtime.controlprogram.parfor.util.IDHandler;
@@ -505,6 +506,9 @@
 		sb.append(DMLScript.getUUID());
 		String dirSuffix = sb.toString();
 		
+		//0) cleanup federated workers if necessary
+		FederatedData.clearFederatedWorkers();
+		
 		//1) cleanup scratch space (everything for current uuid) 
 		//(required otherwise export to hdfs would skip assumed unnecessary writes if same name)
 		HDFSTool.deleteFileIfExistOnHDFS( config.getTextValue(DMLConfig.SCRATCH_SPACE) + dirSuffix );
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java b/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
index c6d7e6e..d5ef760 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
@@ -422,6 +422,10 @@
 		LOG.trace("PARFOR: ParForProgramBlock created with mode = "+_execMode+", optmode = "+_optMode+", numThreads = "+_numThreads);
 	}
 	
+	public static void resetWorkerIDs() {
+		_pwIDSeq.reset();
+	}
+	
 	public long getID() {
 		return _ID;
 	}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/ExecutionContextMap.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/ExecutionContextMap.java
index 1d06f46..352728d 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/ExecutionContextMap.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/ExecutionContextMap.java
@@ -19,6 +19,7 @@
 
 package org.apache.sysds.runtime.controlprogram.federated;
 
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -44,6 +45,18 @@
 			k -> deriveExecutionContext(_main));
 	}
 	
+	public void clear() {
+		//handle main symbol table (w/ tmp list for concurrent modification)
+		for( String varName : new ArrayList<>(_main.getVariables().keySet()) )
+			_main.cleanupDataObject(_main.removeVariable(varName));
+		
+		//handle parfor execution contexts
+		for( ExecutionContext ec : _parEc.values() )
+			for( String varName : ec.getVariables().keySet() )
+				_main.cleanupDataObject(ec.removeVariable(varName));
+		_parEc.clear();
+	}
+	
 	private static ExecutionContext createExecutionContext() {
 		ExecutionContext ec = ExecutionContextFactory.createContext();
 		ec.setAutoCreateVars(true); //w/o createvar inst
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
index 296e6f2..d161522 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
@@ -32,30 +32,37 @@
 import io.netty.handler.codec.serialization.ObjectDecoder;
 import io.netty.handler.codec.serialization.ObjectEncoder;
 import io.netty.util.concurrent.Promise;
+
 import org.apache.sysds.common.Types;
 import org.apache.sysds.conf.DMLConfig;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.RequestType;
 
 import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
 import java.util.concurrent.Future;
 
 
 public class FederatedData {
-	private Types.DataType _dataType;
-	private InetSocketAddress _address;
-	private String _filepath;
+	private static Set<InetSocketAddress> _allFedSites = new HashSet<>();
+	
+	private final Types.DataType _dataType;
+	private final InetSocketAddress _address;
+	private final String _filepath;
 	/**
 	 * The ID of default matrix/tensor on which operations get executed if no other ID is given.
 	 */
 	private long _varID = -1; // -1 is never valid since varIDs start at 0
-	private int _nrThreads = DMLConfig.DEFAULT_NUMBER_OF_FEDERATED_WORKER_THREADS;
-
 
 	public FederatedData(Types.DataType dataType, InetSocketAddress address, String filepath) {
 		_dataType = dataType;
 		_address = address;
 		_filepath = filepath;
+		if( _address != null )
+			_allFedSites.add(_address);
 	}
 	
 	/**
@@ -105,16 +112,20 @@
 		return executeFederatedOperation(request);
 	}
 	
+	public synchronized Future<FederatedResponse> executeFederatedOperation(FederatedRequest... request) {
+		return executeFederatedOperation(_address, request);
+	}
+	
 	/**
 	 * Executes an federated operation on a federated worker.
 	 *
 	 * @param request the requested operation
 	 * @return the response
 	 */
-	public synchronized Future<FederatedResponse> executeFederatedOperation(FederatedRequest... request) {
+	public static Future<FederatedResponse> executeFederatedOperation(InetSocketAddress address, FederatedRequest... request) {
 		// Careful with the number of threads. Each thread opens connections to multiple files making resulting in 
 		// java.io.IOException: Too many open files
-		EventLoopGroup workerGroup = new NioEventLoopGroup(_nrThreads);
+		EventLoopGroup workerGroup = new NioEventLoopGroup(DMLConfig.DEFAULT_NUMBER_OF_FEDERATED_WORKER_THREADS);
 		try {
 			Bootstrap b = new Bootstrap();
 			final DataRequestHandler handler = new DataRequestHandler(workerGroup);
@@ -128,7 +139,7 @@
 				}
 			});
 			
-			ChannelFuture f = b.connect(_address).sync();
+			ChannelFuture f = b.connect(address).sync();
 			Promise<FederatedResponse> promise = f.channel().eventLoop().newPromise();
 			handler.setPromise(promise);
 			f.channel().writeAndFlush(request);
@@ -142,6 +153,21 @@
 		}
 	}
 	
+	public static void clearFederatedWorkers() {
+		if( _allFedSites.isEmpty() )
+			return;
+		
+		//create and execute clear request on all workers
+		FederatedRequest fr = new FederatedRequest(RequestType.CLEAR);
+		List<Future<FederatedResponse>> ret = new ArrayList<>();
+		for( InetSocketAddress address : _allFedSites )
+			ret.add(executeFederatedOperation(address, fr));
+		
+		//wait for successful completion
+		FederationUtils.waitFor(ret);
+		_allFedSites.clear();
+	}
+	
 	private static class DataRequestHandler extends ChannelInboundHandlerAdapter {
 		private Promise<FederatedResponse> _prom;
 		private EventLoopGroup _workerGroup;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java
index d62e6f6..6c9be16 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedRequest.java
@@ -37,6 +37,7 @@
 		GET_VAR,   // return local variable to main
 		EXEC_INST, // execute arbitrary instruction over
 		EXEC_UDF,  // execute arbitrary user-defined function
+		CLEAR,     // clear all variables and execution contexts (i.e., rmvar ALL)
 	}
 	
 	private RequestType _method;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java
index 35b844d..6dd0abc 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java
@@ -119,6 +119,8 @@
 					return execInstruction(request);
 				case EXEC_UDF:
 					return execUDF(request);
+				case CLEAR:
+					return execClear();
 				default:
 					String message = String.format("Method %s is not supported.", method);
 					return new FederatedResponse(ResponseType.ERROR,
@@ -278,6 +280,11 @@
 		}
 	}
 
+	private FederatedResponse execClear() {
+		_ecm.clear();
+		return new FederatedResponse(ResponseType.SUCCESS_EMPTY);
+	}
+	
 	private static void checkNumParams(int actual, int... expected) {
 		if (Arrays.stream(expected).anyMatch(x -> x == actual))
 			return;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java
index b25f8b9..b272bf9 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java
@@ -154,7 +154,7 @@
 		// prepare results (future federated responses), with optional wait to ensure the 
 		// order of requests without data dependencies (e.g., cleanup RPCs)
 		if( wait )
-			waitFor(ret);
+			FederationUtils.waitFor(ret);
 		return ret.toArray(new Future[0]);
 	}
 	
@@ -178,17 +178,7 @@
 		for(FederatedData fd : _fedMap.values())
 			tmp.add(fd.executeFederatedOperation(request));
 		//wait to avoid interference w/ following requests
-		waitFor(tmp);
-	}
-	
-	private static void waitFor(List<Future<FederatedResponse>> responses) {
-		try {
-			for(Future<FederatedResponse> fr : responses)
-				fr.get();
-		}
-		catch(Exception ex) {
-			throw new DMLRuntimeException(ex);
-		}
+		FederationUtils.waitFor(tmp);
 	}
 	
 	private static FederatedRequest[] addAll(FederatedRequest a, FederatedRequest[] b) {
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java
index 429834b..9a14aba 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java
@@ -21,6 +21,7 @@
 
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.Future;
 
 import org.apache.sysds.common.Types.ExecType;
@@ -44,6 +45,10 @@
 public class FederationUtils {
 	private static final IDSequence _idSeq = new IDSequence();
 	
+	public static void resetFedDataID() {
+		_idSeq.reset();
+	}
+	
 	public static long getNextFedDataID() {
 		return _idSeq.getNextID();
 	}
@@ -159,4 +164,14 @@
 			throw new DMLRuntimeException("Unsupported aggregation operator: "
 				+ aop.aggOp.increOp.fn.getClass().getSimpleName());
 	}
+	
+	public static void waitFor(List<Future<FederatedResponse>> responses) {
+		try {
+			for(Future<FederatedResponse> fr : responses)
+				fr.get();
+		}
+		catch(Exception ex) {
+			throw new DMLRuntimeException(ex);
+		}
+	}
 }
diff --git a/src/test/java/org/apache/sysds/test/functions/federated/FederatedKmeansTest.java b/src/test/java/org/apache/sysds/test/functions/federated/FederatedKmeansTest.java
index 7da40ab..7b60476 100644
--- a/src/test/java/org/apache/sysds/test/functions/federated/FederatedKmeansTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/federated/FederatedKmeansTest.java
@@ -26,6 +26,8 @@
 
 import org.apache.sysds.common.Types;
 import org.apache.sysds.common.Types.ExecMode;
+import org.apache.sysds.runtime.controlprogram.ParForProgramBlock;
+import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
 import org.apache.sysds.runtime.meta.MatrixCharacteristics;
 import org.apache.sysds.runtime.util.HDFSTool;
 import org.apache.sysds.test.AutomatedTestBase;
@@ -50,6 +52,8 @@
 	public int cols;
 	@Parameterized.Parameter(2)
 	public int runs;
+	@Parameterized.Parameter(3)
+	public int rep;
 
 	@Override
 	public void setUp() {
@@ -61,8 +65,9 @@
 	public static Collection<Object[]> data() {
 		// rows have to be even and > 1
 		return Arrays.asList(new Object[][] {
-			{10000, 10, 1}, {2000, 50, 1}, {1000, 100, 1},
-			{10000, 10, 2}, {2000, 50, 2}, {1000, 100, 2}, //concurrent requests
+			{10000, 10, 1, 1}, {2000, 50, 1, 1}, {1000, 100, 1, 1},
+			{10000, 10, 2, 1}, {2000, 50, 2, 1}, {1000, 100, 2, 1}, //concurrent requests
+			{10000, 10, 2, 2}, //repeated exec
 			//TODO more runs e.g., 16 -> but requires rework RPC framework first
 			//(e.g., see paramserv?)
 		});
@@ -115,26 +120,31 @@
 			"-nvargs", "in_X1=" + TestUtils.federatedAddress(port1, input("X1")),
 			"in_X2=" + TestUtils.federatedAddress(port2, input("X2")), "rows=" + rows, "cols=" + cols,
 			"runs=" + String.valueOf(runs), "out=" + output("Z")};
-		runTest(true, false, null, -1);
-
+		
+		for( int i=0; i<rep; i++ ) {
+			ParForProgramBlock.resetWorkerIDs();
+			FederationUtils.resetFedDataID();
+			runTest(true, false, null, -1);
+		
+			// check for federated operations
+			Assert.assertTrue(heavyHittersContainsString("fed_ba+*"));
+			Assert.assertTrue(heavyHittersContainsString("fed_uasqk+"));
+			Assert.assertTrue(heavyHittersContainsString("fed_uarmin"));
+			Assert.assertTrue(heavyHittersContainsString("fed_*"));
+			Assert.assertTrue(heavyHittersContainsString("fed_+"));
+			Assert.assertTrue(heavyHittersContainsString("fed_<="));
+			Assert.assertTrue(heavyHittersContainsString("fed_/"));
+			Assert.assertTrue(heavyHittersContainsString("fed_r'"));
+			
+			//check that federated input files are still existing
+			Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X1")));
+			Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X2")));
+		}
+		
 		// compare via files
 		//compareResults(1e-9); --> randomized
 		TestUtils.shutdownThreads(t1, t2);
 		
-		// check for federated operations
-		Assert.assertTrue(heavyHittersContainsString("fed_ba+*"));
-		Assert.assertTrue(heavyHittersContainsString("fed_uasqk+"));
-		Assert.assertTrue(heavyHittersContainsString("fed_uarmin"));
-		Assert.assertTrue(heavyHittersContainsString("fed_*"));
-		Assert.assertTrue(heavyHittersContainsString("fed_+"));
-		Assert.assertTrue(heavyHittersContainsString("fed_<="));
-		Assert.assertTrue(heavyHittersContainsString("fed_/"));
-		Assert.assertTrue(heavyHittersContainsString("fed_r'"));
-		
-		//check that federated input files are still existing
-		Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X1")));
-		Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X2")));
-		
 		resetExecMode(platformOld);
 	}
 }
diff --git a/src/test/java/org/apache/sysds/test/functions/federated/FederatedPCATest.java b/src/test/java/org/apache/sysds/test/functions/federated/FederatedPCATest.java
index 906b124..4b4457a 100644
--- a/src/test/java/org/apache/sysds/test/functions/federated/FederatedPCATest.java
+++ b/src/test/java/org/apache/sysds/test/functions/federated/FederatedPCATest.java
@@ -83,19 +83,28 @@
 		String HOME = SCRIPT_DIR + TEST_DIR;
 
 		// write input matrices
-		int halfRows = rows / 2;
+		int quarterRows = rows / 4;
 		// We have two matrices handled by a single federated worker
-		double[][] X1 = getRandomMatrix(halfRows, cols, 0, 1, 1, 3);
-		double[][] X2 = getRandomMatrix(halfRows, cols, 0, 1, 1, 7);
-		writeInputMatrixWithMTD("X1", X1, false, new MatrixCharacteristics(halfRows, cols, blocksize, halfRows * cols));
-		writeInputMatrixWithMTD("X2", X2, false, new MatrixCharacteristics(halfRows, cols, blocksize, halfRows * cols));
+		double[][] X1 = getRandomMatrix(quarterRows, cols, 0, 1, 1, 3);
+		double[][] X2 = getRandomMatrix(quarterRows, cols, 0, 1, 1, 7);
+		double[][] X3 = getRandomMatrix(quarterRows, cols, 0, 1, 1, 8);
+		double[][] X4 = getRandomMatrix(quarterRows, cols, 0, 1, 1, 9);
+		MatrixCharacteristics mc= new MatrixCharacteristics(quarterRows, cols, blocksize, quarterRows * cols);
+		writeInputMatrixWithMTD("X1", X1, false, mc);
+		writeInputMatrixWithMTD("X2", X2, false, mc);
+		writeInputMatrixWithMTD("X3", X3, false, mc);
+		writeInputMatrixWithMTD("X4", X4, false, mc);
 
 		// empty script name because we don't execute any script, just start the worker
 		fullDMLScriptName = "";
 		int port1 = getRandomAvailablePort();
 		int port2 = getRandomAvailablePort();
+		int port3 = getRandomAvailablePort();
+		int port4 = getRandomAvailablePort();
 		Thread t1 = startLocalFedWorker(port1);
 		Thread t2 = startLocalFedWorker(port2);
+		Thread t3 = startLocalFedWorker(port3);
+		Thread t4 = startLocalFedWorker(port4);
 
 		TestConfiguration config = availableTestConfigurations.get(TEST_NAME);
 		loadTestConfiguration(config);
@@ -103,21 +112,24 @@
 		
 		// Run reference dml script with normal matrix
 		fullDMLScriptName = HOME + TEST_NAME + "Reference.dml";
-		programArgs = new String[] {"-stats", "-args", input("X1"), input("X2"),
+		programArgs = new String[] {"-stats", "-args", input("X1"), input("X2"), input("X3"), input("X4"),
 			String.valueOf(scaleAndShift).toUpperCase(), expected("Z")};
 		runTest(true, false, null, -1);
 
 		// Run actual dml script with federated matrix
 		fullDMLScriptName = HOME + TEST_NAME + ".dml";
-		programArgs = new String[] {"-stats",
-			"-nvargs", "in_X1=" + TestUtils.federatedAddress(port1, input("X1")),
-			"in_X2=" + TestUtils.federatedAddress(port2, input("X2")), "rows=" + rows, "cols=" + cols,
+		programArgs = new String[] {"-stats", "-nvargs", 
+			"in_X1=" + TestUtils.federatedAddress(port1, input("X1")),
+			"in_X2=" + TestUtils.federatedAddress(port2, input("X2")),
+			"in_X3=" + TestUtils.federatedAddress(port3, input("X3")),
+			"in_X4=" + TestUtils.federatedAddress(port4, input("X4")),
+			"rows=" + rows, "cols=" + cols,
 			"scaleAndShift=" + String.valueOf(scaleAndShift).toUpperCase(), "out=" + output("Z")};
 		runTest(true, false, null, -1);
 
 		// compare via files
 		compareResults(1e-9);
-		TestUtils.shutdownThreads(t1, t2);
+		TestUtils.shutdownThreads(t1, t2, t3, t4);
 		
 		// check for federated operations
 		Assert.assertTrue(heavyHittersContainsString("fed_ba+*"));
@@ -134,6 +146,8 @@
 		//check that federated input files are still existing
 		Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X1")));
 		Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X2")));
+		Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X3")));
+		Assert.assertTrue(HDFSTool.existsFileOnHDFS(input("X4")));
 		
 		resetExecMode(platformOld);
 	}
diff --git a/src/test/scripts/functions/federated/FederatedPCATest.dml b/src/test/scripts/functions/federated/FederatedPCATest.dml
index b235d44..049a789 100644
--- a/src/test/scripts/functions/federated/FederatedPCATest.dml
+++ b/src/test/scripts/functions/federated/FederatedPCATest.dml
@@ -19,7 +19,8 @@
 #
 #-------------------------------------------------------------
 
-X = federated(addresses=list($in_X1, $in_X2),
-    ranges=list(list(0, 0), list($rows / 2, $cols), list($rows / 2, 0), list($rows, $cols)))
+X = federated(addresses=list($in_X1, $in_X2, $in_X3, $in_X4),
+    ranges=list(list(0, 0), list($rows/4, $cols), list($rows/4, 0), list(2*$rows/4, $cols),
+		list(2*$rows/4, 0), list(3*$rows/4, $cols), list(3*$rows/4, 0), list($rows, $cols)))
 [X2,M] = pca(X=X,  K=2, scale=$scaleAndShift, center=$scaleAndShift)
 write(X2, $out)
diff --git a/src/test/scripts/functions/federated/FederatedPCATestReference.dml b/src/test/scripts/functions/federated/FederatedPCATestReference.dml
index 0b17fe0..3ab3b97 100644
--- a/src/test/scripts/functions/federated/FederatedPCATestReference.dml
+++ b/src/test/scripts/functions/federated/FederatedPCATestReference.dml
@@ -19,6 +19,6 @@
 #
 #-------------------------------------------------------------
 
-X = rbind(read($1), read($2))
-[X2,M] = pca(X=X,  K=2, scale=$3, center=$3)
-write(X2, $4)
+X = rbind(read($1), read($2), read($3), read($4));
+[X2,M] = pca(X=X,  K=2, scale=$5, center=$5)
+write(X2, $6)