Merge branch 'main' into sbak/fr_fix

Author: sbak5

src/nvidia_resiliency_ext/fault_tolerance/config.py

@@ -229,10 +229,6 @@ def from_args(args: argparse.Namespace):
         ft_cfg._fix_log_level_type()
         ft_cfg._fix_rank_termination_signal_type()
 
-        # If we didn't read from file and no CLI args were provided, raise an error
-        if not (is_read_from_file or cli_ft_args):
-            raise ValueError("No fault tolerance configuration provided.")
-
         return ft_cfg
 
     def to_yaml_file(self, cfg_path: str) -> None:

src/nvidia_resiliency_ext/fault_tolerance/launcher.py

@@ -408,18 +408,22 @@ def _invoke_run_with_any_failed_policy(self, role: str = DEFAULT_ROLE) -> RunRes
                         rank=self._worker_group.group_rank,
                     )
 
-                    logger.info(
-                        "[%s] Detected cluster changes from group_rank=%s "
-                        "(unhealthy_nodes=%s, nodes_waiting=%s); will restart worker group",
-                        role,
-                        group_rank,
-                        unhealthy_count,
-                        num_nodes_waiting,
-                    )
-
-                    # Note: The node that triggered the change (unhealthy or new) already opened
-                    # the rendezvous, so we don't need to open it again here.
-                    self._restart_workers(self._worker_group)
+                    if self._remaining_restarts > 0:
+                        logger.info(
+                            "[%s] Detected cluster changes from group_rank=%s "
+                            "(unhealthy_nodes=%s, nodes_waiting=%s); will restart worker group",
+                            role,
+                            group_rank,
+                            unhealthy_count,
+                            num_nodes_waiting,
+                        )
+                        self._remaining_restarts -= 1
+                        # Note: The node that triggered the change (unhealthy or new) already opened
+                        # the rendezvous, so we don't need to open it again here.
+                        self._restart_workers(self._worker_group)
+                    else:
+                        self._stop_workers(self._worker_group)
+                        return RunResult(state=WorkerState.FAILED)
             else:
                 raise Exception(f"[{role}] Worker group in {state.name} state")
 
@@ -1983,12 +1987,12 @@ def get_args_parser() -> ArgumentParser:
         "--ft-rdzv_impl",
         type=str,
         choices=["barrier", "legacy"],
-        default="legacy",
+        default="barrier",
         dest="ft_rdzv_impl",
         help="FT rendezvous implementation to use. "
         "'barrier' uses the new atomic barrier-based algorithm (ft_rendezvous_barrier.py), "
         "'legacy' uses the original compare-and-set algorithm (_ft_rendezvous.py). "
-        "Default: legacy. Note: This is independent of --rdzv-backend (which specifies "
+        "Default: barrier. Note: This is independent of --rdzv-backend (which specifies "
         "the coordination backend like c10d or etcd).",
     )