[MINOR] Extended DML config for federated worker parallelism

sysds.federated.par_conn (number of concurrent connections in even loop)
sysds.federated.par_inst (instruction parallelism worker instructions)

For both, if the value is <=0, we use by default the number of virtual
cores as reported by the JVM.

Author: mboehm7

conf/SystemDS-config.xml.template

@@ -118,6 +118,12 @@
     <!-- set the federated plan generator (none, [runtime], compile_fed_all, compile_fed_heuristic, compile_cost_based) -->
     <sysds.federated.planner>runtime</sysds.federated.planner>
 
+    <!-- set the degree of parallelism of the federated worker event loop (<=0 means number of virtual cores) -->
+    <sysds.federated.par_conn>0</sysds.federated.par_conn>
+
+    <!-- set the degree of parallelism of the federated worker instructions (<=0 means number of virtual cores) -->
+    <sysds.federated.par_inst>0</sysds.federated.par_inst>
+
     <!-- set buffer pool threshold (max size) in % of total heap -->
     <sysds.caching.bufferpoollimit>15</sysds.caching.bufferpoollimit>
 

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

@@ -116,6 +116,8 @@ public class DMLConfig
 	public static final String DEFAULT_FEDERATED_INITIALIZATION_TIMEOUT = "sysds.federated.initialization.timeout"; // int seconds
 	public static final String FEDERATED_TIMEOUT = "sysds.federated.timeout"; // single request timeout default -1 to indicate infinite.
 	public static final String FEDERATED_PLANNER = "sysds.federated.planner";
+	public static final String FEDERATED_PAR_INST = "sysds.federated.par_inst";
+	public static final String FEDERATED_PAR_CONN = "sysds.federated.par_conn";
 	public static final int DEFAULT_FEDERATED_PORT = 4040; // borrowed default Spark Port
 	public static final int DEFAULT_NUMBER_OF_FEDERATED_WORKER_THREADS = 2;
 
@@ -181,6 +183,8 @@ public class DMLConfig
 		_defaultVals.put(DEFAULT_FEDERATED_INITIALIZATION_TIMEOUT, "10");
 		_defaultVals.put(FEDERATED_TIMEOUT,      "-1");
 		_defaultVals.put(FEDERATED_PLANNER,      FederatedPlanner.RUNTIME.name());
+		_defaultVals.put(FEDERATED_PAR_CONN,     "-1"); // vcores
+		_defaultVals.put(FEDERATED_PAR_INST,     "-1"); // vcores
 	}
 
 	public DMLConfig() {

src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDCZero.java

@@ -28,6 +28,8 @@
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 
 public abstract class ASDCZero extends APreAgg {
+	private static final long serialVersionUID = -69266306137398807L;
+	
 	/** Sparse row indexes for the data */
 	protected AOffset _indexes;
 

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

@@ -44,6 +44,7 @@
 import org.apache.log4j.Logger;
 import org.apache.sysds.conf.ConfigurationManager;
 import org.apache.sysds.conf.DMLConfig;
+import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
 import org.apache.sysds.runtime.lineage.LineageCacheConfig;
 
 public class FederatedWorker {
@@ -67,7 +68,8 @@ public FederatedWorker(int port, boolean debug) {
 
 	public void run() throws CertificateException, SSLException {
 		log.info("Setting up Federated Worker on port " + _port);
-		final int EVENT_LOOP_THREADS = Math.max(4, Runtime.getRuntime().availableProcessors() * 4);
+		int par_conn = ConfigurationManager.getDMLConfig().getIntValue(DMLConfig.FEDERATED_PAR_CONN);
+		final int EVENT_LOOP_THREADS = (par_conn > 0) ? par_conn : InfrastructureAnalyzer.getLocalParallelism();
 		NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
 		ThreadPoolExecutor workerTPE = new ThreadPoolExecutor(1, Integer.MAX_VALUE,
 			10, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(true));

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

@@ -37,6 +37,7 @@
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.FileFormat;
 import org.apache.sysds.conf.ConfigurationManager;
+import org.apache.sysds.conf.DMLConfig;
 import org.apache.sysds.parser.DataExpression;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.controlprogram.BasicProgramBlock;
@@ -48,6 +49,7 @@
 import org.apache.sysds.runtime.controlprogram.context.SparkExecutionContext;
 import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.RequestType;
 import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse.ResponseType;
+import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
 import org.apache.sysds.runtime.instructions.Instruction;
 import org.apache.sysds.runtime.instructions.Instruction.IType;
 import org.apache.sysds.runtime.instructions.InstructionParser;
@@ -436,8 +438,9 @@ private FederatedResponse execInstruction(FederatedRequest request, ExecutionCon
 
 		// set the number of threads according to the number of processors on the federated worker
 		if(receivedInstruction.getOperator() instanceof MultiThreadedOperator) {
-			int numProcessors = Runtime.getRuntime().availableProcessors();
-			((MultiThreadedOperator)receivedInstruction.getOperator()).setNumThreads(numProcessors);
+			int par_inst = ConfigurationManager.getDMLConfig().getIntValue(DMLConfig.FEDERATED_PAR_INST);
+			((MultiThreadedOperator)receivedInstruction.getOperator())
+				.setNumThreads((par_inst > 0) ? par_inst : InfrastructureAnalyzer.getLocalParallelism());
 		}
 
 		BasicProgramBlock pb = new BasicProgramBlock(null);