which function sends information to the aggregation function?

[dskhanirfan]

what is the name of the function that is run on the participating collaborators which sends information to the aggregation function? Can you outline the function calls in a sequential manner i-e one function call after the other that leads to final aggregation in an FL round?

[sarthakpati]

Tagging @psfoley

[dskhanirfan]

In order to trace function calls in FeTS_Challenge.py file, I used the following trace module

python -m trace --listfuncs FeTS_Challenge.py

Output

Requirement already satisfied: torchvision in /projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages (from -r /users/irfan21h/.local/workspace/requirements.txt (line 1)) (0.9.1)
Requirement already satisfied: torch in /projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages (from -r /users/irfan21h/.local/workspace/requirements.txt (line 2)) (1.8.1)
Requirement already satisfied: numpy in /appl/soft/bio/bioconda/miniconda3/lib/python3.7/site-packages (from torchvision->-r /users/irfan21h/.local/workspace/requirements.txt (line 1)) (1.21.2)
Requirement already satisfied: pillow>=4.1.1 in /appl/soft/bio/bioconda/miniconda3/lib/python3.7/site-packages (from torchvision->-r /users/irfan21h/.local/workspace/requirements.txt (line 1)) (8.3.2)
Requirement already satisfied: typing-extensions in /appl/soft/bio/bioconda/miniconda3/lib/python3.7/site-packages (from torch->-r /users/irfan21h/.local/workspace/requirements.txt (line 2)) (3.10.0.0)
No 'TrainOrVal' column found in split_subdirs csv, so performing automated split using percent_train of 0.8
[23:35:25] INFO     Updating aggregator.settings.rounds_to_train to 50...                                                                           native.py:83
[23:35:29] INFO     Updating aggregator.settings.db_store_rounds to 50...                                                                           native.py:83
/appl/soft/bio/bioconda/miniconda3/lib/python3.7/site-packages/pandas/core/frame.py:4913: SettingWithCopyWarning: 
A value is trying to be set on a copy of a slice from a DataFrame

See the caveats in the documentation: https://pandas.pydata.org/pandas-docs/stable/user_guide/indexing.html#returning-a-view-versus-a-copy
  errors=errors,
Traceback (most recent call last):
  File "/appl/soft/bio/bioconda/miniconda3/lib/python3.7/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/appl/soft/bio/bioconda/miniconda3/lib/python3.7/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/appl/soft/bio/bioconda/miniconda3/lib/python3.7/trace.py", line 724, in <module>
    main()
  File "/appl/soft/bio/bioconda/miniconda3/lib/python3.7/trace.py", line 712, in main
    t.runctx(code, globs, globs)
  File "/appl/soft/bio/bioconda/miniconda3/lib/python3.7/trace.py", line 449, in runctx
    exec(cmd, globals, locals)
  File "DP_SimAgg.py", line 863, in <module>
    restore_from_checkpoint_folder = restore_from_checkpoint_folder)
  File "/scratch/project_2004598/07JAN2022/Challenge/Task_1/fets_challenge/experiment.py", line 289, in run_challenge_experiment
    task_runner = copy(plan).get_task_runner(list(collaborator_data_loaders.values())[0])
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/openfl/federated/plan/plan.py", line 340, in get_task_runner
    self.runner_ = Plan.Build(**defaults)
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/openfl/federated/plan/plan.py", line 180, in Build
    instance = getattr(module, class_name)(**settings)
  File "/users/irfan21h/.local/workspace/impl/fets_challenge_model.py", line 108, in __init__
    self.__init_2(val_input_shape=val_input_shape, val_output_shape=val_output_shape, data=data, **kwargs)
  File "/users/irfan21h/.local/workspace/impl/fets_challenge_model.py", line 319, in __init_2
    self.channel_keys= self.get_channel_keys()
  File "/users/irfan21h/.local/workspace/impl/fets_challenge_model.py", line 343, in get_channel_keys
    for subject in self.data.get_train_loader():
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/dataloader.py", line 517, in __next__
    data = self._next_data()
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/dataloader.py", line 557, in _next_data
    data = self._dataset_fetcher.fetch(index)  # may raise StopIteration
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/_utils/fetch.py", line 44, in fetch
    data = [self.dataset[idx] for idx in possibly_batched_index]
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/_utils/fetch.py", line 44, in <listcomp>
    data = [self.dataset[idx] for idx in possibly_batched_index]
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/data/queue.py", line 164, in __getitem__
    self.fill()
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/data/queue.py", line 228, in fill
    subject = self.get_next_subject()
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/data/queue.py", line 238, in get_next_subject
    subject = next(self.subjects_iterable)
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/dataloader.py", line 517, in __next__
    data = self._next_data()
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/dataloader.py", line 557, in _next_data
    data = self._dataset_fetcher.fetch(index)  # may raise StopIteration
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/_utils/fetch.py", line 44, in fetch
    data = [self.dataset[idx] for idx in possibly_batched_index]
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torch/utils/data/_utils/fetch.py", line 44, in <listcomp>
    data = [self.dataset[idx] for idx in possibly_batched_index]
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/data/dataset.py", line 85, in __getitem__
    subject = self._transform(subject)
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/transforms/transform.py", line 121, in __call__
    transformed = self.apply_transform(subject)
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/transforms/augmentation/composition.py", line 47, in apply_transform
    subject = transform(subject)
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/transforms/transform.py", line 121, in __call__
    transformed = self.apply_transform(subject)
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/torchio/transforms/lambda_transform.py", line 54, in apply_transform
    result = self.function(function_arg)
  File "/projappl/project_2004598/.py_venv_3.7.3/lib/python3.7/site-packages/GANDLF/preprocessing.py", line 76, in tensor_rotate_90
    return torch.transpose(input_image, affected_axes[0], affected_axes[1]).flip(affected_axes[1])

Which function performs segmentation on participating collaborators and sends parameters to the aggregator function?

[psfoley]

Hi @dskhanirfan, your high level understanding in the other thread was correct.

Here is the function call sequence on the collaborator:

run()
--> Here the collaborator continually invokes a gRPC request to the aggregator (get_tasks) to see which tasks it should perform for that round. The aggregator will return the list of tasks (to be performed in order), and from there the collaborator will invoke its do_call function
do_call()
----> This function performs the following:

1. Looks up in the plan.yaml configuration what FeTSChallengeModel function that task name refers to, and which kwargs it passes in for that round. For the aggregated_model_validation task, this function would resolve the validate function from the FeTSChallengeModel and pass in kwargs global and custom metrics.
2. The necessary TensorKeys for that function are resolved in order for that function to run. First, the collaborator will look in its own TensorDB for the entries, and if they're not available it will query the aggregator for the necessary tensors ( get_aggregated_tensor RPC). For aggregated_model_validation the required tensors are the aggregated model from the end of the last round.
3. After the required numpy arrays are resolved, the FeTSChallengeModel function is finally invoked (i.e. validate in the case of the aggregated_model_validation task). The validate function updates its model with the newly passed in numpy arrays, performs validation, and then returns as output the dictionary of {TensorKey: numpy array} that correspond to the metrics intended to go back to the aggregator.
4. Now returning back to the do_task function after the invocation of the FeTSChallengeModel's validate function, the collaborator sends the TensorKeys and corresponding numpy arrays for the aggregated_model_validation task to the aggregator through using the send_local_task_results RPC. The aggregator will store these in its TensorDB, and will aggregate them after all tasks by all collaborators have been completed.
5. After that task is completed, do_call() returns, and is invoked again if there is another task to be performed

At the aggregator, when the last task by the last collaborator is performed, the _is_round_done() function will return true, at which point the model tensors and training/validation metrics will be aggregated, any custom aggregation functions defined as part of challenge will be called (get_aggregated_tensor), and metrics for the round will be logged and the latest model is saved). After this, the model for the next round is ready to download by each of the collaborators, and the next round starts.