[SYSTEMDS-3845] Federated Threading Bug

The federated back end spawn threads for parallel execution instead of using the threadpool.
This commit fixes the issue by naming the worker threads to enable the threadpool usage.

The performance on a local experiment using the FederatedKMeans test improve from 4.3 sec average to 3.2 sec average. To reproduce the results set the federated k-means test to repeat the federated call 20 times.

Closes #2245

Author: Baunsgaard

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

@@ -616,6 +616,10 @@ private static void exec(ExecutionContext ec, Instruction ins){
 
 		try {
 			// execute single instruction
+			// TODO move this thread naming to Netty thread creation!
+			Thread curThread = Thread.currentThread();
+			long id = curThread.getId();
+			Thread.currentThread().setName("FedExec_"+ id);
 			pb.execute(ec);
 		}
 		catch(Exception ex) {

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

@@ -115,7 +115,7 @@ public synchronized static ExecutorService get(int k) {
 		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")) {
+		else if(mainThread || threadName.contains("PARFOR") || threadName.contains("FedExec")) {
 			CommonThreadPool pool;
 			if(shared2 == null) // If there is no current shared pool allocate one.
 				shared2 = new ConcurrentHashMap<>();

src/test/java/org/apache/sysds/test/functions/federated/algorithms/FederatedKmeansTest.java

@@ -22,6 +22,8 @@
 import java.util.Arrays;
 import java.util.Collection;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.sysds.common.Types;
 import org.apache.sysds.common.Types.ExecMode;
 import org.apache.sysds.runtime.controlprogram.ParForProgramBlock;
@@ -31,6 +33,7 @@
 import org.apache.sysds.test.AutomatedTestBase;
 import org.apache.sysds.test.TestConfiguration;
 import org.apache.sysds.test.TestUtils;
+import org.apache.sysds.utils.stats.Timing;
 import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -40,6 +43,7 @@
 @RunWith(value = Parameterized.class)
 @net.jcip.annotations.NotThreadSafe
 public class FederatedKmeansTest extends AutomatedTestBase {
+	protected static final Log LOG = LogFactory.getLog(FederatedKmeansTest.class.getName());
 
 	private final static String TEST_DIR = "functions/federated/";
 	private final static String TEST_NAME = "FederatedKmeansTest";
@@ -120,7 +124,6 @@ public void federatedKmeans(Types.ExecMode execMode, boolean singleWorker) {
 		programArgs = new String[] {"-args", input("X1"), input("X2"),
 			String.valueOf(singleWorker).toUpperCase(), String.valueOf(runs), expected("Z")};
 		runTest(true, false, null, -1);
-
 		// Run actual dml script with federated matrix
 		fullDMLScriptName = HOME + TEST_NAME + ".dml";
 		programArgs = new String[] {"-stats","20", "-nvargs", "in_X1=" + TestUtils.federatedAddress(port1, input("X1")),
@@ -130,8 +133,9 @@ public void federatedKmeans(Types.ExecMode execMode, boolean singleWorker) {
 		for(int i = 0; i < rep; i++) {
 			ParForProgramBlock.resetWorkerIDs();
 			FederationUtils.resetFedDataID();
+			Timing t = new Timing();
 			runTest(true, false, null, -1);
-
+			LOG.debug("Federated kmeans runtime: " + t);
 			// check for federated operations
 			Assert.assertTrue(heavyHittersContainsString("fed_ba+*"));
 			// Assert.assertTrue(heavyHittersContainsString("fed_uasqk+"));