Feature/plan agreement

openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Portland/Portland_config.yaml

@@ -1,6 +1,7 @@
 settings:
   director_host: localhost
   director_port: 50050
+  review_experiment: True
 
 Portland:
   private_attributes: private_attributes.portland_attrs

openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/Seattle/Seattle_config.yaml

@@ -1,6 +1,7 @@
 settings:
   director_host: localhost
   director_port: 50050
+  review_experiment: True  
 
 Seattle:
   private_attributes: private_attributes.seattle_attrs

openfl-tutorials/experimental/workflow/FederatedRuntime/101_MNIST/director/director_config.yaml

@@ -1,4 +1,5 @@
 settings:
   listen_host: localhost
   listen_port: 50050
-  envoy_health_check_period: 5  # in seconds
+  envoy_health_check_period: 5  # in seconds
+  review_experiment: True

openfl/experimental/workflow/component/aggregator/aggregator.py

@@ -220,6 +220,10 @@ async def run_flow(self) -> FLSpec:
                     logger.info(f"Tasks will not be sent to {k}")
 
             while not self.collaborator_task_results.is_set():
+                # Check if its time to quite 
+                if self.time_to_quit:
+                    logger.info("Time to quit detected. Exiting wait loop")
+                    break
                 len_sel_collabs = len(self.selected_collaborators)
                 len_connected_collabs = len(self.connected_collaborators)
                 if len_connected_collabs < len_sel_collabs:
@@ -243,13 +247,16 @@ async def run_flow(self) -> FLSpec:
                         + " collaborators to send results..."
                     )
                 await asyncio.sleep(Aggregator._get_sleep_time())
+
+            if self.time_to_quit:
+                logger.info("Time to quit detected after waiting loop . Stopping aggregator")
+                break
 
             self.collaborator_task_results.clear()
             f_name = self.next_step
             if hasattr(self, "instance_snapshot"):
                 self.flow.restore_instance_snapshot(self.flow, list(self.instance_snapshot))
                 delattr(self, "instance_snapshot")
-
         return self.flow
 
     def call_checkpoint(
@@ -309,25 +316,32 @@ def get_tasks(self, collaborator_name: str) -> Tuple:
             f"Aggregator GetTasks function reached from collaborator {collaborator_name}..."
         )
 
+
+        # Check if time_to_quit is set by aggregator . 
+        # If set, send quit job to collaborator and return
+        if self.time_to_quit:
+            return self._send_shutdown_signal(collaborator_name)
+
         # If queue of requesting collaborator is empty
         while self.__collaborator_tasks_queue[collaborator_name].qsize() == 0:
             # If it is time to then inform the collaborator
             if self.time_to_quit:
-                logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...")
-                self.quit_job_sent_to.append(collaborator_name)
-                # FIXME: 0, and "" instead of None is just for protobuf compatibility.
-                #  Cleaner solution?
-                return (
-                    0,
-                    "",
-                    None,
-                    Aggregator._get_sleep_time(),
-                    self.time_to_quit,
-                )
-
+                return self._send_shutdown_signal(collaborator_name)
+
             # If not time to quit then sleep for 10 seconds
             time.sleep(Aggregator._get_sleep_time())
 
+        '''
+        # Check if quit job is sent to collaborator
+        if collaborator_name in self.quit_job_sent_to:
+            logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...")
+            return (0, "", None, Aggregator._get_sleep_time(), self.time_to_quit)
+        # Check if quit job is sent to all collaborators
+        if self.all_quit_jobs_sent():
+            logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...")
+            return (0, "", None, Aggregator._get_sleep_time(), self.time_to_quit)
+        '''
+
         # Get collaborator step, and clone for requesting collaborator
         next_step, clone = self.__collaborator_tasks_queue[collaborator_name].get()
 
@@ -343,6 +357,14 @@ def get_tasks(self, collaborator_name: str) -> Tuple:
             0,
             self.time_to_quit,
         )
+
+    def _send_shutdown_signal(self, collaborator_name: str) -> Tuple:
+        """Send a shutdown signal to the collaborator."""
+        if collaborator_name not in self.quit_job_sent_to:
+            logger.info(f"Sending signal to collaborator {collaborator_name} to shutdown...")  
+            self.quit_job_sent_to.append(collaborator_name)
+        return (0, "", None, Aggregator._get_sleep_time(), self.time_to_quit)
+
 
     def do_task(self, f_name: str) -> Any:
         """Execute aggregator steps until transition.
@@ -514,7 +536,55 @@ def valid_collaborator_cn_and_id(
 
     def all_quit_jobs_sent(self) -> bool:
         """Assert all quit jobs are sent to collaborators."""
+        #logger.info(f"self.quit_job_sent_to {self.quit_job_sent_to} ,self.authorized_cols {self.authorized_cols}") # Added for debugging
+        # Check if all quit jobs are sent to collaborators
         return set(self.quit_job_sent_to) == set(self.authorized_cols)
+
+    '''
+    def stop(self, failed_collaborator: str = None) -> None:
+        """Stop the aggregator and all collaborators.
+
+        Args:
+            failed_collaborator (str, optional): Name of the failed
+                collaborator. Defaults to None.
+        """
+        if failed_collaborator is not None:
+            logger.info(f"Stopping aggregator due to failure of {failed_collaborator}.")
+        else:
+            logger.info("Stopping aggregator.")
+        self.time_to_quit = True
+        self.collaborator_task_results.set()
+        # Stop all collaborators
+        for collab in self.authorized_cols:
+            if collab not in self.quit_job_sent_to:
+                logger.info(f"Sending quit job to {collab}.")
+                self.quit_job_sent_to.append(collab)
+    ''' 
+    def stop(self, failed_collaborator: str = None) -> None:
+        """Stop the aggregator , failed collaborator and send quit jobs to all collaborators.
+
+        Args:
+            failed_collaborator (str, optional): Name of the failed
+                collaborator. Defaults to None.
+        """
+        logger.info("Force stopping the aggregator execution.")
+        self.time_to_quit = True  # Signal All collaborators to quit
+
+        if failed_collaborator:
+            logger.info(f"Stopping aggregator triggred by {failed_collaborator}.")
+            if failed_collaborator not in self.quit_job_sent_to:
+                logger.info(f"Sending quit job to {failed_collaborator}.")
+                self.quit_job_sent_to.append(failed_collaborator)
+
+        # Send quit jobs to all collaborators.
+        #for collaborator_name in self.authorized_cols:
+            #if collaborator_name not in self.quit_job_sent_to:
+                #logger.info(f"Sending quit job to collaborator '{collaborator_name}'.")
+                #self.quit_job_sent_to.append(collaborator_name)
+
+
+
+
 
 
 the_dragon = """

openfl/experimental/workflow/component/collaborator/collaborator.py

@@ -138,6 +138,8 @@ def run(self) -> None:
         while True:
             next_step, clone, sleep_time, time_to_quit = self.get_tasks()
             if time_to_quit:
+                self.logger.info(f"Collaborator {self.name} received Quit signal. ⛔Shutting down.....")
+                self.logger.info("Received end of federation signal. Exiting...")
                 break
             elif sleep_time > 0:
                 time.sleep(sleep_time)

openfl/experimental/workflow/component/director/director.py

@@ -18,6 +18,7 @@
     ExperimentsRegistry,
 )
 from openfl.experimental.workflow.transport.grpc.exceptions import EnvoyNotFoundError
+from openfl.experimental.workflow.component.director.experiment import Status
 
 logger = logging.getLogger(__name__)
 
@@ -45,6 +46,7 @@ class Director:
         envoy_health_check_period (int): The period for health check of envoys
             in seconds.
         authorized_cols (list): A list of authorized envoys
+        review_callback (Optional[Callable]): A callback function for reviewing experiments.
     """
 
     def __init__(
@@ -57,6 +59,7 @@ def __init__(
         director_config: Optional[Path] = None,
         envoy_health_check_period: int = 60,
         install_requirements: bool = True,
+        review_callback = None,  # Add review_callback parameter
     ) -> None:
         """Initialize a Director object.
 
@@ -74,6 +77,7 @@ def __init__(
             in seconds.
             install_requirements (bool, optional): A flag indicating if the
                 requirements should be installed. Defaults to True.
+            review_callback (Optional[Callable]): A callback function for reviewing experiments.
         """
         self.tls = tls
         self.root_certificate = root_certificate
@@ -82,7 +86,7 @@ def __init__(
         self.director_config = director_config
         self.install_requirements = install_requirements
         self._flow_status = asyncio.Queue()
-
+        self.review_callback = review_callback  # Store the review_callback
         self.experiments_registry = ExperimentsRegistry()
         self.col_exp = {}
         self.col_exp_queues = defaultdict(asyncio.Queue)
@@ -172,16 +176,16 @@ async def set_new_experiment(
         collaborator_names: Iterable[str],
         experiment_archive_path: Path,
     ) -> bool:
-        """Set new experiment.
+        """Set new experiment and optionally review experiment .
 
         Args:
-            experiment_name (str): String id for experiment.
-            sender_name (str): The name of the sender.
-            collaborator_names (Iterable[str]): Names of collaborators.
-            experiment_archive_path (Path): Path of the experiment.
+            experiment_name (str): Identifier for the new experiment.
+            sender_name (str): Initiator of the experiment.
+            collaborator_names (Iterable[str]): Participating collaborators.
+            experiment_archive_path (Path): Path to the experiment archive.
 
         Returns:
-            bool : Boolean returned if the experiment register was successful.
+            bool: True if the experiment is accepted and registered; False otherwise.
         """
         experiment = Experiment(
             name=experiment_name,
@@ -190,10 +194,18 @@ async def set_new_experiment(
             users=[sender_name],
             sender=sender_name,
         )
-
+        # Check if review callback is enabled
+        if self.review_callback:
+            review_approved = await experiment.review_experiment(self.review_callback)
+            if not review_approved:
+                logger.warning(f"Experiment '{experiment_name}' was rejected❌ by the Director Admin during review.")
+                return False # Experiment rejected
+
+        # Add the experiment to the registry
         self.authorized_cols = collaborator_names
         self.experiments_registry.add(experiment)
-        return True
+        logger.info(f"Experiment '{experiment_name}' was approved✅ and added to the registry.")
+        return True # Experiment approved
 
     async def stream_experiment_stdout(
         self, experiment_name: str, caller: str
@@ -234,6 +246,7 @@ async def stream_experiment_stdout(
                 # Yield none if the queue is empty but the experiment is still running.
                 yield None
 
+
     def get_experiment_data(self, experiment_name: str) -> Path:
         """Get experiment data.
 
@@ -315,3 +328,30 @@ def update_envoy_status(
         )
 
         return self.envoy_health_check_period
+
+
+    def set_experiment_failed(self, *, experiment_name: str, collaborator_name: str):
+        """ Handle experiment Failure triggred by a collaborator.
+
+        Args:
+            experiment_name (str): String id for experiment.
+            collaborator_name (str): Name of the collaborator.
+        """
+        if experiment_name not in self.experiments_registry:
+            logger.warning(f"Experiment '{experiment_name}' not found in registry.")
+            return
+        experiment = self.experiments_registry[experiment_name]
+        aggregator = experiment.aggregator
+
+        if aggregator:
+            logger.info(f"Stopping aggregator for experiment '{experiment_name}' due to failure by collaborator '{collaborator_name}'.")
+            aggregator.stop(failed_collaborator=collaborator_name)
+
+        experiment.status = Status.FAILED
+        logger.info(f"Experiment '{experiment_name}' marked as FAILED.")
+
+
+
+
+
+