Make choice of aggregator dynamic at session level and additional aggregators (FedOpt)

fedn/fedn/network/combiner/aggregators/aggregatorbase.py

@@ -19,18 +19,18 @@ class AggregatorBase(ABC):
     :type server: class: `fedn.network.combiner.Combiner`
     :param modelservice: A handle to the model service :class: `fedn.network.combiner.modelservice.ModelService`
     :type modelservice: class: `fedn.network.combiner.modelservice.ModelService`
-    :param control: A handle to the :class: `fedn.network.combiner.round.RoundController`
-    :type control: class: `fedn.network.combiner.round.RoundController`
+    :param control: A handle to the :class: `fedn.network.combiner.round.RoundHandler`
+    :type control: class: `fedn.network.combiner.round.RoundHandler`
     """
 
     @abstractmethod
-    def __init__(self, storage, server, modelservice, control):
+    def __init__(self, storage, server, modelservice, round_handler):
         """ Initialize the aggregator."""
         self.name = self.__class__.__name__
         self.storage = storage
         self.server = server
         self.modelservice = modelservice
-        self.control = control
+        self.round_handler = round_handler
         self.model_updates = queue.Queue()
 
     @abstractmethod
@@ -58,7 +58,7 @@ def on_model_update(self, model_update):
         and then puts the update id on the aggregation queue.
         Override in subclass as needed.
 
-        :param model_update: A ModelUpdate message.
+        :param model_update: fedn.network.grpc.fedn.proto.ModelUpdate
         :type model_id: str
         """
         try:
@@ -114,7 +114,7 @@ def load_model_update(self, model_update, helper):
         :rtype: tuple
         """
         model_id = model_update.model_update_id
-        model = self.control.load_model_update(helper, model_id)
+        model = self.round_handler.load_model_update(helper, model_id)
         # Get relevant metadata
         data = json.loads(model_update.meta)['training_metadata']
         config = json.loads(json.loads(model_update.meta)['config'])
@@ -141,8 +141,8 @@ def get_aggregator(aggregator_module_name, storage, server, modelservice, contro
     :type server: class: `fedn.network.combiner.Combiner`
     :param modelservice: A handle to the model service :class: `fedn.network.combiner.modelservice.ModelService`
     :type modelservice: class: `fedn.network.combiner.modelservice.ModelService`
-    :param control: A handle to the :class: `fedn.network.combiner.round.RoundController`
-    :type control: class: `fedn.network.combiner.round.RoundController`
+    :param control: A handle to the :class: `fedn.network.combiner.round.RoundHandler`
+    :type control: class: `fedn.network.combiner.round.RoundHandler`
     :return: An aggregator instance.
     :rtype: class: `fedn.combiner.aggregators.AggregatorBase`
     """

fedn/fedn/network/combiner/aggregators/fedavg.py

@@ -14,15 +14,15 @@ class Aggregator(AggregatorBase):
     :type server: class: `fedn.network.combiner.Combiner`
     :param modelservice: A handle to the model service :class: `fedn.network.combiner.modelservice.ModelService`
     :type modelservice: class: `fedn.network.combiner.modelservice.ModelService`
-    :param control: A handle to the :class: `fedn.network.combiner.round.RoundController`
-    :type control: class: `fedn.network.combiner.round.RoundController`
+    :param control: A handle to the :class: `fedn.network.combiner.round.RoundHandler`
+    :type control: class: `fedn.network.combiner.round.RoundHandler`
 
     """
 
-    def __init__(self, storage, server, modelservice, control):
+    def __init__(self, storage, server, modelservice, round_handler):
         """Constructor method"""
 
-        super().__init__(storage, server, modelservice, control)
+        super().__init__(storage, server, modelservice, round_handler)
 
         self.name = "fedavg"
 

fedn/fedn/network/combiner/aggregators/fedopt.py

@@ -20,14 +20,14 @@ class Aggregator(AggregatorBase):
     :type server: class: `fedn.network.combiner.Combiner`
     :param modelservice: A handle to the model service :class: `fedn.network.combiner.modelservice.ModelService`
     :type modelservice: class: `fedn.network.combiner.modelservice.ModelService`
-    :param control: A handle to the :class: `fedn.network.combiner.round.RoundController`
-    :type control: class: `fedn.network.combiner.round.RoundController`
+    :param control: A handle to the :class: `fedn.network.combiner.round.RoundHandler`
+    :type control: class: `fedn.network.combiner.round.RoundHandler`
 
     """
 
-    def __init__(self, storage, server, modelservice, control):
+    def __init__(self, storage, server, modelservice, round_handler):
 
-        super().__init__(storage, server, modelservice, control)
+        super().__init__(storage, server, modelservice, round_handler)
 
         self.name = "fedopt"
         self.v = None
@@ -81,7 +81,7 @@ def combine_models(self, helper=None, delete_models=True):
                 total_examples += metadata['num_examples']
 
                 if nr_aggregated_models == 0:
-                    model_old = self.control.load_model_update(helper, model_update.model_id)
+                    model_old = self.round_handler.load_model_update(helper, model_update.model_id)
                     pseudo_gradient = helper.subtract(model_next, model_old)
                 else:
                     pseudo_gradient_next = helper.subtract(model_next, model_old)

fedn/fedn/network/combiner/combiner.py

@@ -16,7 +16,7 @@
                                     set_log_stream)
 from fedn.network.combiner.connect import ConnectorCombiner, Status
 from fedn.network.combiner.modelservice import ModelService
-from fedn.network.combiner.round import RoundController
+from fedn.network.combiner.roundhandler import RoundHandler
 from fedn.network.grpc.server import Server
 from fedn.network.storage.s3.repository import Repository
 from fedn.network.storage.statestore.mongostatestore import MongoStateStore
@@ -134,10 +134,10 @@ def __init__(self, config):
         self.server = Server(self, self.modelservice, grpc_config)
 
         # Set up round controller
-        self.control = RoundController(self.repository, self, self.modelservice)
+        self.round_handler = RoundHandler(self.repository, self, self.modelservice)
 
         # Start thread for round controller
-        threading.Thread(target=self.control.run, daemon=True).start()
+        threading.Thread(target=self.round_handler.run, daemon=True).start()
 
         # Start thread for client status updates: TODO: Should be configurable
         threading.Thread(target=self._deamon_thread_client_status, daemon=True).start()
@@ -385,7 +385,7 @@ def _flush_model_update_queue(self):
         :return: True if successful, else False
         """
 
-        q = self.control.aggregator.model_updates
+        q = self.round_handler.aggregator.model_updates
         try:
             with q.mutex:
                 q.queue.clear()
@@ -418,7 +418,7 @@ def Start(self, control: fedn.ControlRequest, context):
 
         logger.debug("grpc.Combiner.Start: Round config {}".format(config))
 
-        job_id = self.control.push_round_config(config)
+        job_id = self.round_handler.push_round_config(config)
         logger.info("grcp.Combiner.Start: Pushed round config (job_id): {}".format(job_id))
 
         response = fedn.ControlResponse()
@@ -442,7 +442,7 @@ def SetAggregator(self, control: fedn.ControlRequest, context):
         for parameter in control.parameter:
             aggregator = parameter.value
 
-        status = self.control.set_aggregator(aggregator)
+        status = self.round_handler.set_aggregator(aggregator)
 
         response = fedn.ControlResponse()
         if status:
@@ -723,7 +723,7 @@ def SendModelUpdate(self, request, context):
         :return: the response
         :rtype: :class:`fedn.network.grpc.fedn_pb2.Response`
         """
-        self.control.aggregator.on_model_update(request)
+        self.round_handler.aggregator.on_model_update(request)
 
         response = fedn.Response()
         response.response = "RECEIVED ModelUpdate {} from client  {}".format(

fedn/fedn/network/combiner/round.py → fedn/fedn/network/combiner/roundhandler.py

@@ -13,11 +13,11 @@ class ModelUpdateError(Exception):
     pass
 
 
-class RoundController:
-    """ Round controller.
+class RoundHandler:
+    """ Round handler.
 
-    The round controller recieves round configurations from the global controller
-    and coordinates model updates and aggregation, and model validations.
+    The round handler processes requests from the global controller
+    to produce model updates and perform model validations.
 
     :param aggregator_name: The name of the aggregator plugin module.
     :type aggregator_name: str
@@ -30,7 +30,7 @@ class RoundController:
     """
 
     def __init__(self, storage, server, modelservice):
-        """ Initialize the RoundController."""
+        """ Initialize the RoundHandler."""
 
         self.round_configs = queue.Queue()
         self.storage = storage
@@ -58,9 +58,9 @@ def push_round_config(self, round_config):
         return round_config['_job_id']
 
     def load_model_update(self, helper, model_id):
-        """Load model update in its native format.
+        """Load model update with id model_id into its memory representation.
 
-        :param helper: An instance of :class: `fedn.utils.helpers.helpers.HelperBase`, ML framework specific helper, defaults to None
+        :param helper: An instance of :class: `fedn.utils.helpers.helpers.HelperBase`
         :type helper: class: `fedn.utils.helpers.helpers.HelperBase`
         :param model_id: The ID of the model update, UUID in str format
         :type model_id: str