[GitHub] [systemds] tobiasrieger commented on a change in pull request #1141: [SYSTEMDS-2550] Batch scaling and weighing of imbalanced workers

[GitBox <gi...@apache.org>]

tobiasrieger commented on a change in pull request #1141:
URL: https://github.com/apache/systemds/pull/1141#discussion_r551866215



##########
File path: src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/FederatedPSControlThread.java
##########
@@ -58,48 +61,68 @@
 public class FederatedPSControlThread extends PSWorker implements Callable<Void> {
 	private static final long serialVersionUID = 6846648059569648791L;
 	protected static final Log LOG = LogFactory.getLog(ParamServer.class.getName());
-	
-	Statement.PSRuntimeBalancing _runtimeBalancing;
+
 	FederatedData _featuresData;
 	FederatedData _labelsData;
 	final long _localStartBatchNumVarID;
 	final long _modelVarID;
-	int _numBatchesPerGlobalEpoch;
+
+	// runtime balancing
+	Statement.PSRuntimeBalancing _runtimeBalancing;
+	int _numBatchesPerEpoch;
 	int _possibleBatchesPerLocalEpoch;
+	boolean _weighing;
+	double _weighingFactor = 1;
 	boolean _cycleStartAt0 = false;
 
-	public FederatedPSControlThread(int workerID, String updFunc, Statement.PSFrequency freq, Statement.PSRuntimeBalancing runtimeBalancing, int epochs, long batchSize, int numBatchesPerGlobalEpoch, ExecutionContext ec, ParamServer ps) {
+	public FederatedPSControlThread(int workerID, String updFunc, Statement.PSFrequency freq,
+									Statement.PSRuntimeBalancing runtimeBalancing, boolean weighing, int epochs, long batchSize,
+									int numBatchesPerGlobalEpoch, ExecutionContext ec, ParamServer ps) {
 		super(workerID, updFunc, freq, epochs, batchSize, ec, ps);
 
-		_numBatchesPerGlobalEpoch = numBatchesPerGlobalEpoch;
+		_numBatchesPerEpoch = numBatchesPerGlobalEpoch;
 		_runtimeBalancing = runtimeBalancing;
+		_weighing = weighing;
 		// generate the IDs for model and batch counter. These get overwritten on the federated worker each time
 		_localStartBatchNumVarID = FederationUtils.getNextFedDataID();
 		_modelVarID = FederationUtils.getNextFedDataID();
 	}
 
 	/**
 	 * Sets up the federated worker and control thread
+	 *
+	 * @param weighingFactor Gradients from this worker will be multiplied by this factor if weighing is enabled
 	 */
-	public void setup() {
+	public void setup(double weighingFactor) {
 		// prepare features and labels
 		_featuresData = (FederatedData) _features.getFedMapping().getMap().values().toArray()[0];
 		_labelsData = (FederatedData) _labels.getFedMapping().getMap().values().toArray()[0];
 
-		// calculate number of batches and get data size
+		// weighing factor is always set, but only used when weighing is specified
+		_weighingFactor = weighingFactor;
+
+		// different runtime balancing calculations
 		long dataSize = _features.getNumRows();
-		_possibleBatchesPerLocalEpoch = (int) Math.ceil((double) dataSize / _batchSize);
-		if(!(_runtimeBalancing == Statement.PSRuntimeBalancing.RUN_MIN 
-			|| _runtimeBalancing == Statement.PSRuntimeBalancing.CYCLE_AVG 
-			|| _runtimeBalancing == Statement.PSRuntimeBalancing.CYCLE_MAX)) {
-			_numBatchesPerGlobalEpoch = _possibleBatchesPerLocalEpoch;
+
+		// calculate scaled batch size if balancing via batch size.
+		// In some cases there will be some cycling
+		if(_runtimeBalancing == Statement.PSRuntimeBalancing.SCALE_BATCH) {
+			_batchSize = (int) Math.ceil((float) dataSize / _numBatchesPerEpoch);
 		}
 
-		if(_runtimeBalancing == Statement.PSRuntimeBalancing.SCALE_BATCH 
-			|| _runtimeBalancing == Statement.PSRuntimeBalancing.SCALE_BATCH_AND_WEIGH) {
-			throw new NotImplementedException();
+		// Calculate possible batches with batch size
+		_possibleBatchesPerLocalEpoch = (int) Math.ceil((double) dataSize / _batchSize);
+
+		// If no runtime balancing is specified, just run possible number of batches
+		// WARNING: Will get stuck on miss match
+		if(_runtimeBalancing == Statement.PSRuntimeBalancing.NONE) {
+			_numBatchesPerEpoch = _possibleBatchesPerLocalEpoch;

Review comment:
       Thanks for taking a look! A missmatch occurs when you specify a batch size and two workers have a different number of examples. For Example batchsize: 1 Worker 1: 20 and Worker 2: 30 examples. Without data partitioning or runtime balancing Worker 2 will wait for Worker 1 to complete batch 21 which it never does therefor getting stuck. I'll think about adding a detection feature and an exception in the next PR!




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org