Processing logic of controller

[jingyunc]

Hello, I'm looking for the code piece in NVFlare where the processing logic of a controller is defined.

I'm trying to build my own controller, following the example of swarm_client_ctl.py. I saw classes and methods were defined inside the example py, but didn't find a "main" function.

In the Programming Guide of Swarm Learning, I saw written description of the processing logic: "The workflow is started from the starting_client..." I wonder where these logics were implemented in the codes, as I need to modify them for my own controller.

Thanks in advance!

[chesterxgchen]

@SYangster

[SYangster]

Hi, great question! There isn't a "main" function in the swarm_client_ctl.py, as controllers (as well as other components) are all used within the running FLARE system. A main function would be called in the specific case of using our new ClientAPI to launch a subprocess, however the Swarm Learning example does not currently use this.

For the processing logic "The workflow is started from the starting_client..." if you look at
swarm_client_ctl.py(353), there is a start_workflow() function that is called and kicks off the scatter logic from the starting client for swarm learning. This start_workflow() is executed by the ClientSideController when it receives start_task_name task from the ServerSideController.

Since Swarm Learning uses a Client-Controlled Workflow rather than a Server-Controlled Workflow, the ServerSideController in this case is just managing the job lifecycle, and the start_task_name is just telling the ClientSideController to start it's workflow. As such, for modifying your own controller, start_workflow() would be a good place to start looking at to change.

Hope this helps!

[jingyunc]

Thanks @SYangster for the clarification. I have a further question about swarm_client_ctl.py:

At each round, after the training clients send their results to the aggregator, if I want add a step: to validate each incoming model on aggregator's local data (to dynamically set its weight in the aggregation), how may I to do that?

This is an important step of my FL algorithm, but I struggled to implement it in terms of NVFlare controller.

Any suggestion would be helpful.

[SYangster]

Sure, here's some rough ideas that might help:

1. Where in the workflow to add this validation step: After the clients have sent their results to the aggregator, which would be after the gather step and right before the actual aggregation (swarm_client_ctl.py:162).
2. Pass incoming models for validation: Need a modified aggregator to obtain the accepted models. Also currently the InTimeAccumulateWeightedAggregator is configured for "expected_data_kind": "WEIGHT_DIFF" when receiving the result from the CIFAR10ModelLearner train task, so that needs to be changed to either apply the weight diffs or change to use weights)
3. Validate each incoming model on aggregator’s local dataset: Given that we are using a ModelLearner that already supports the validate task, we can do something similar to (swarm_client_ctl.py:604) result = self.execute_learn_task(task_data, fl_ctx, abort_signal). Instead of the learn_executor executing self.learn_task_name which is “train”, we write a new function to execute “validate” with the task_data being the accepted model from a client.
4. Dynamically set the aggregation weights: Need to modify aggregator to set this based on the validation results instead of how it’s currently handled in the constructor.

Let me know if you have any more questions or need clarification!

[jingyunc]

Thanks @SYangster, great to know this implementation is possible in NVFlare. I have some follow-up questions:

For 2: can you specify how to gather the accepted models (for validation) in the aggregator? I came across following codes from the hello_cyclic example, which converts model weights to learnable objects:

dxo = from_shareable(shareable)
model_weights = dxo.data

These codes were used in trainer.py. May I use them directly in the aggregator.py?

For 3: where in the intime_accumulate_model_aggregator.py should I insert the line result = self.execute_validate_task(...)? Any classes I should import in order to execute learner tasks in the aggregator? I understand I can save the modified aggregator to a separate file and link it in configure.

Also for 3: how may I tell the execute_validate_task() to run on the data of aggregator site, rather than contributor site?

Thanks in advance.

[SYangster]

2: Currently, in InTimeAccumulateWeightedAggregrator, the accept() function is used to obtain the results from the clients (swarm_client_ctl.py 149). Here you have the right idea of getting the model weights from the dxo. This current accept() function receives the shareable and extracts the dxo, however right now it stores the accepted dxo in self.dxo_aggregrators. In your case you would want to store the dxo in a new data structure in the aggregator that can be accessed when you want to do the validation step.

3: For where to execute the validate task, we would want to do this in the controller, not the aggregator (component), as controllers are responsible for sending tasks and receiving their results. One such place to call this would be after all the results have been accepted (swarm_client_ctl.py:161), and you could loop through the clients’ models that you stored in (2).

In order to ensure that the validate task runs on the data of the aggregator site, we can leverage the check self.me == aggr (swarm_client_ctl.py:572). Because the self.gatherer is set only when this condition is true, and executor=self is passed into this Gatherer, we have all we need in order to call execute_validate_task() on the aggregator site. Calling execute_validate_task() from the Gatherer in the place described above (3) would involve something similar to self.executor.learn_executor.execute(“validate”, client_weights_from_your_aggregrator, fl_ctx, abort_signal)

[jingyunc]

It worked, many thanks! I have a less related question, so will post in a separate thread.