[Feature] Upgrade node roles architecture and SDFL (#47)

* feature RoleBehavior interface

* fix cfl and trust role behavior implementation

* mod minor changes on disconnection method

* updt minor changes

* update:
- get_round concurrence access
- update shutdown protocol

* feature model propagation event

* upgrade shutdown protocol
update controller network remove

* upgrade shutdown process on node

* clean code

* fake behavior added

* feature leadership tranfer

* add trainer_aggregator role

* fix removing nodes

* trainer aggregator role added

* feature leadership transfer among nodes

* fix self selection on leadership transfer

* feature resolver no updates conflict

* upgrade documentation

* upgrade aggregation waiting time release

* upgrade monitor SDFL

* remove duplicates and comments

* latest reputation changes added

* get_round fixed

---------

Co-authored-by: FerTV <fernando.torres.vega@gmail.com>
Co-authored-by: FerTV <45948619+FerTV@users.noreply.github.com>

Author: AlejandroAvilesSerrano

nebula/addons/gps/nebulagps.py

@@ -90,7 +90,7 @@ async def _send_location_loop(self):
                 from nebula.core.network.communications import CommunicationsManager
 
                 cm = CommunicationsManager.get_instance()
-                if cm.learning_finished():
+                if await cm.learning_finished():
                     logging.info("GPS: Learning cycle finished, stopping location broadcast")
                     break
             except Exception:
@@ -111,7 +111,7 @@ async def _receive_location_loop(self):
                 from nebula.core.network.communications import CommunicationsManager
 
                 cm = CommunicationsManager.get_instance()
-                if cm.learning_finished():
+                if await cm.learning_finished():
                     logging.info("GPS: Learning cycle finished, stopping location reception")
                     break
             except Exception:
@@ -139,7 +139,7 @@ async def _notify_geolocs(self):
                 from nebula.core.network.communications import CommunicationsManager
 
                 cm = CommunicationsManager.get_instance()
-                if cm.learning_finished():
+                if await cm.learning_finished():
                     logging.info("GPS: Learning cycle finished, stopping geolocation notifications")
                     break
             except Exception:

nebula/addons/mobility.py

@@ -77,20 +77,20 @@ def __init__(self, config, verbose=False):
     def cm(self):
         return CommunicationsManager.get_instance()
 
-    @property
-    def round(self):
-        """
-        Gets the current round number from the Communications Manager.
-
-        This property retrieves the current round number that is being managed by the
-        CommunicationsManager instance associated with this module. It provides an
-        interface to access the ongoing round of the communication process without
-        directly exposing the underlying method in the CommunicationsManager.
-
-        Returns:
-            int: The current round number managed by the CommunicationsManager.
-        """
-        return self.cm.get_round()
+    # @property
+    # def round(self):
+    #     """
+    #     Gets the current round number from the Communications Manager.
+
+    #     This property retrieves the current round number that is being managed by the
+    #     CommunicationsManager instance associated with this module. It provides an
+    #     interface to access the ongoing round of the communication process without
+    #     directly exposing the underlying method in the CommunicationsManager.
+
+    #     Returns:
+    #         int: The current round number managed by the CommunicationsManager.
+    #     """
+    #     return self.cm.get_round()
 
     async def start(self):
         """

nebula/addons/networksimulation/nebulanetworksimulator.py

@@ -51,41 +51,40 @@ async def is_running(self):
 
     async def _change_network_conditions_based_on_distances(self, gpsevent: GPSEvent):
         distances = await gpsevent.get_event_data()
-        while await self.is_running():
-            if self._verbose:
-                logging.info("Refresh | conditions based on distances...")
-            try:
-                for addr, (distance, _) in distances.items():
-                    if distance is None:
-                        # If the distance is not found, we skip the node
-                        continue
-                    conditions = await self._calculate_network_conditions(distance)
-                    # Only update the network conditions if they have changed
-                    if (
-                        addr not in self._current_network_conditions
-                        or self._current_network_conditions[addr] != conditions
-                    ):
-                        addr_ip = addr.split(":")[0]
-                        self._set_network_condition_for_addr(
-                            self._node_interface, addr_ip, conditions["bandwidth"], conditions["delay"]
-                        )
-                        self._set_network_condition_for_multicast(
-                            self._node_interface,
-                            addr_ip,
-                            self.IP_MULTICAST,
-                            conditions["bandwidth"],
-                            conditions["delay"],
-                        )
-                        async with self._network_conditions_lock:
-                            self._current_network_conditions[addr] = conditions
-                    else:
-                        if self._verbose:
-                            logging.info("network conditions havent changed since last time")
-            except KeyError:
-                logging.exception(f"📍  Connection {addr} not found")
-            except Exception:
-                logging.exception("📍  Error changing connections based on distance")
-            await asyncio.sleep(self._refresh_interval)
+        if self._verbose:
+            logging.info("Refresh | conditions based on distances...")
+        try:
+            for addr, (distance, _) in distances.items():
+                if distance is None:
+                    # If the distance is not found, we skip the node
+                    continue
+                conditions = await self._calculate_network_conditions(distance)
+                # Only update the network conditions if they have changed
+                if (
+                    addr not in self._current_network_conditions
+                    or self._current_network_conditions[addr] != conditions
+                ):
+                    addr_ip = addr.split(":")[0]
+                    self._set_network_condition_for_addr(
+                        self._node_interface, addr_ip, conditions["bandwidth"], conditions["delay"]
+                    )
+                    self._set_network_condition_for_multicast(
+                        self._node_interface,
+                        addr_ip,
+                        self.IP_MULTICAST,
+                        conditions["bandwidth"],
+                        conditions["delay"],
+                    )
+                    async with self._network_conditions_lock:
+                        self._current_network_conditions[addr] = conditions
+                else:
+                    if self._verbose:
+                        logging.info("network conditions havent changed since last time")
+        except KeyError:
+            logging.exception(f"📍  Connection {addr} not found")
+        except Exception:
+            logging.exception("📍  Error changing connections based on distance")
+        await asyncio.sleep(self._refresh_interval)
 
     async def set_thresholds(self, thresholds: dict):
         async with self._network_conditions_lock: