From c5ce9d82766a9472e1b935bab4ac597ff98185ca Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 9 Jun 2022 15:30:31 +0000 Subject: [PATCH 01/27] proof of concept for elastic training using pytorch --- deepspeed/constants.py | 4 +- deepspeed/launcher/launch.py | 122 +++++++++++++++++++++++++++++++---- 2 files changed, 110 insertions(+), 16 deletions(-) diff --git a/deepspeed/constants.py b/deepspeed/constants.py index 9576c9c078d5..318d2461bfe4 100644 --- a/deepspeed/constants.py +++ b/deepspeed/constants.py @@ -13,7 +13,7 @@ # (only if NCCL_BLOCKING_WAIT or NCCL_ASYNC_ERROR_HANDLING is set to 1). # To make an attempt at backwards compatibility with THD, we use an # extraordinarily high default timeout, given that THD did not have timeouts. -default_pg_timeout = timedelta(minutes=30) - +# default_pg_timeout = timedelta(minutes=30) +default_pg_timeout = timedelta(seconds=30) INFERENCE_GENERIC_MODE = 'generic' INFERENCE_SPECIALIZED_MODE = 'specialized' diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 21ef04b338b7..435bfeb04ab7 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -17,12 +17,19 @@ import signal from collections import defaultdict from argparse import ArgumentParser, REMAINDER - +from torch.distributed.launcher.api import LaunchConfig, elastic_launch +from torch.distributed.elastic.multiprocessing import Std from ..constants import TORCH_DISTRIBUTED_DEFAULT_PORT from ..utils import logger - PID_FILE_BASEPATH = "/tmp" - +sys.path.append('/home/t-arpanjain/work/elastic_try/using_torch/') +from functionAgent import FunctionElasticAgent +from torch.distributed.elastic.rendezvous import RendezvousParameters +from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent +import torch.distributed.elastic.rendezvous.registry as rdzv_registry +from torch.distributed.elastic.agent.server.api import WorkerSpec +from torch.distributed.elastic.multiprocessing import Std +from typing import Any, Dict, Optional, Tuple def parse_args(): parser = ArgumentParser(description="DeepSpeed distributed training launch" @@ -87,6 +94,32 @@ def parse_args(): return parser.parse_args() +def get_config_elastic(args, num_local_procs, node_rank) -> LaunchConfig: + max_nodes = args.nnodes *2 + min_nodes = 1 + config: Dict[str, str] = {'timeout': 100} + # config["rank"] = node_rank + config["store_type "] = "file" + + config = LaunchConfig( + min_nodes=min_nodes, + max_nodes=max_nodes, + nproc_per_node=num_local_procs, + run_id="123456789", + role="default", + rdzv_endpoint="worker-0:46728", + rdzv_backend='c10d', + rdzv_configs=config, + max_restarts=100, + monitor_interval=1, + start_method="spawn", + redirects=Std.from_str("0"), + tee=Std.from_str("0"), + log_dir="", + ) + return config + + def main(): args = parse_args() current_env = os.environ.copy() @@ -145,11 +178,40 @@ def main(): processes = [] cmd = [] - for local_rank in range(0, num_local_procs): - # each process's rank - dist_rank = global_rank_mapping[local_node][local_rank] - current_env["RANK"] = str(dist_rank) - current_env["LOCAL_RANK"] = str(local_rank) + Elastic_training = False + + if not Elastic_training: + for local_rank in range(0, num_local_procs): + # each process's rank + dist_rank = global_rank_mapping[local_node][local_rank] + current_env["RANK"] = str(dist_rank) + current_env["LOCAL_RANK"] = str(local_rank) + + # spawn the processes + cmd = [] + if not args.no_python: + cmd = [sys.executable, "-u"] + if args.module: + cmd.append("-m") + else: + if args.module: + raise ValueError("Don't use both the '--no_python' flag" + " and the '--module' flag at the same time.") + cmd.append(args.training_script) + # A user may not want to pass local_rank as a keyword arg so we make this optional. + if not args.no_local_rank: + cmd.append(f"--local_rank={local_rank}") + cmd += args.training_script_args + + process = subprocess.Popen(cmd, env=current_env) + processes.append(process) + else: + # dist_rank = global_rank_mapping[local_node][local_rank] + # os.environ["RANK"] = str(dist_rank) + # os.environ["LOCAL_RANK"] = str(local_rank) + + os.environ["MASTER_ADDR"] = args.master_addr + os.environ["MASTER_PORT"] = str(args.master_port) # spawn the processes cmd = [] @@ -160,15 +222,47 @@ def main(): else: if args.module: raise ValueError("Don't use both the '--no_python' flag" - " and the '--module' flag at the same time.") + " and the '--module' flag at the same time.") cmd.append(args.training_script) - # A user may not want to pass local_rank as a keyword arg so we make this optional. - if not args.no_local_rank: - cmd.append(f"--local_rank={local_rank}") cmd += args.training_script_args + elastic_config = get_config_elastic(args, num_local_procs,args.node_rank) + cmd_args = cmd[1:] + # cmd_args = ['MASTER_ADDR={}'.format(args.master_addr), 'MASTER_PORT={}'.format(args.master_port)] + cmd_args + print ("CMD is:",cmd_args) + + + # elastic_launch( + # config=elastic_config, + # entrypoint= cmd[0], + # )(*cmd_args) + rdzv_configs: Dict[str, str] = {'timeout': 100} + rdzv_parameters = RendezvousParameters( + backend='c10d', + endpoint="worker-0:29401", + run_id='123456789', + min_nodes=1, + max_nodes=2, + **rdzv_configs + ) + + spec = WorkerSpec( + role='trainer', + local_world_size=8, + entrypoint=cmd[0], + args=cmd[1:], + rdzv_handler=rdzv_registry.get_rendezvous_handler(rdzv_parameters), + max_restarts=100, + monitor_interval=50, + redirects=Std.from_str("0"), + tee=Std.from_str("0"), + master_addr='worker-0', + master_port='51000', + ) + agent = FunctionElasticAgent( + spec + ) + agent.run() - process = subprocess.Popen(cmd, env=current_env) - processes.append(process) sig_names = {2: "SIGINT", 15: "SIGTERM"} last_return_code = None From 5635052a3fa14c152878a4b6fbf8dc55efd3d6d5 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 16 Jun 2022 10:16:44 +0500 Subject: [PATCH 02/27] Add command line options for elastic training --- deepspeed/launcher/launch.py | 39 ++++++++++++++++---------- deepspeed/launcher/multinode_runner.py | 3 ++ deepspeed/launcher/runner.py | 37 +++++++++++++++++++----- 3 files changed, 57 insertions(+), 22 deletions(-) diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 435bfeb04ab7..b8b3ecf3010f 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -71,6 +71,15 @@ def parse_args(): help="Skip prepending the training script with " "'python' - just execute it directly.") + parser.add_argument("--enable_elastic_training", + action="store_true", + help="Enable elastic training support.") + + parser.add_argument("--max_nodes", + type=int, + default=-1, + help="Max number of nodes in elastic training.") + parser.add_argument("--no_local_rank", action="store_true", help="Do not pass local_rank as an argument when calling " @@ -95,15 +104,13 @@ def parse_args(): def get_config_elastic(args, num_local_procs, node_rank) -> LaunchConfig: - max_nodes = args.nnodes *2 - min_nodes = 1 + config: Dict[str, str] = {'timeout': 100} - # config["rank"] = node_rank config["store_type "] = "file" config = LaunchConfig( - min_nodes=min_nodes, - max_nodes=max_nodes, + min_nodes=args.nnodes, + max_nodes=args.max_nodes, nproc_per_node=num_local_procs, run_id="123456789", role="default", @@ -124,6 +131,8 @@ def main(): args = parse_args() current_env = os.environ.copy() + + for k in current_env.keys(): if "NCCL" in k: logger.info(f"{args.node_rank} {k}={current_env[k]}") @@ -178,9 +187,8 @@ def main(): processes = [] cmd = [] - Elastic_training = False - if not Elastic_training: + if not args.enable_elastic_training: for local_rank in range(0, num_local_procs): # each process's rank dist_rank = global_rank_mapping[local_node][local_rank] @@ -209,6 +217,7 @@ def main(): # dist_rank = global_rank_mapping[local_node][local_rank] # os.environ["RANK"] = str(dist_rank) # os.environ["LOCAL_RANK"] = str(local_rank) + assert args.max_nodes > 0, "Max nodes should be provided in elastic training" os.environ["MASTER_ADDR"] = args.master_addr os.environ["MASTER_PORT"] = str(args.master_port) @@ -238,27 +247,27 @@ def main(): rdzv_configs: Dict[str, str] = {'timeout': 100} rdzv_parameters = RendezvousParameters( backend='c10d', - endpoint="worker-0:29401", + endpoint=args.master_addr+":29401", run_id='123456789', - min_nodes=1, - max_nodes=2, + min_nodes=args.nnodes, + max_nodes=args.max_nodes, **rdzv_configs ) spec = WorkerSpec( role='trainer', - local_world_size=8, + local_world_size=num_local_procs, entrypoint=cmd[0], args=cmd[1:], rdzv_handler=rdzv_registry.get_rendezvous_handler(rdzv_parameters), max_restarts=100, - monitor_interval=50, + monitor_interval=5, redirects=Std.from_str("0"), tee=Std.from_str("0"), - master_addr='worker-0', - master_port='51000', + master_addr=args.master_addr, + master_port=str(args.master_port), ) - agent = FunctionElasticAgent( + agent = LocalElasticAgent( spec ) agent.run() diff --git a/deepspeed/launcher/multinode_runner.py b/deepspeed/launcher/multinode_runner.py index a962a8a7c925..2e885bc47c6f 100644 --- a/deepspeed/launcher/multinode_runner.py +++ b/deepspeed/launcher/multinode_runner.py @@ -94,6 +94,9 @@ def get_cmd(self, environment, active_resources): deepspeed_launch.append("--no_local_rank") if self.args.save_pid: deepspeed_launch += ["--save_pid", f"{os.getpid()}"] + if self.args.elastic_training: + deepspeed_launch.append("--enable_elastic_training") + deepspeed_launch.append(f"--max_nodes={self.args.max_num_nodes}") return pdsh_cmd_args + deepspeed_launch + [self.user_script ] + self.user_arguments diff --git a/deepspeed/launcher/runner.py b/deepspeed/launcher/runner.py index be25715660db..df55f9754618 100755 --- a/deepspeed/launcher/runner.py +++ b/deepspeed/launcher/runner.py @@ -70,8 +70,8 @@ def parse_args(args=None): ''') parser.add_argument("--num_nodes", - type=int, - default=-1, + type=str, + default="-1", help="Total number of worker nodes to run on, this will use " "the top N hosts from the given hostfile.") @@ -146,6 +146,10 @@ def parse_args(args=None): help="Run DeepSpeed autotuner to discover optimal configuration parameters " "before running job.") + parser.add_argument("--elastic_training", + action="store_true", + help="Enable elastic training support in DeepSpeed.") + parser.add_argument("user_script", type=str, help="User script to launch, followed by any required " @@ -313,10 +317,26 @@ def run_autotuning(args, active_resources): if args.autotuning == "run": tuner.run_after_tuning() +def parse_num_nodes(str_num_nodes: str, elastic_training: bool): + node_list = str_num_nodes.split(":") + + if len(node_list) == 1: + min_nodes, max_nodes = int(node_list[0]), -1 + elif len(node_list) == 2 and elastic_training: + min_nodes, max_nodes = int(node_list[0]), int(node_list[1]) + elif len(node_list) == 2 and not elastic_training: + raise RuntimeError("MIN:MAX format is only supported in elastic training") + else: + raise RuntimeError("num_nodes {} is not in MIN:MAX format".format(str_num_nodes)) + + return min_nodes, max_nodes + def main(args=None): args = parse_args(args) + num_nodes, args.max_num_nodes = parse_num_nodes(args.num_nodes, args.elastic_training) + resource_pool = fetch_hostfile(args.hostfile) # respect CUDA_VISIBLE_DEVICES for a single node and no explicit resource filters @@ -324,7 +344,7 @@ def main(args=None): if not resource_pool and len(cuda_visible_devices): detected_str = f"Detected CUDA_VISIBLE_DEVICES={cuda_visible_devices}" if len(args.include) or len( - args.exclude) or args.num_nodes > 1 or args.num_gpus > 0: + args.exclude) or num_nodes > 1 or args.num_gpus > 0: print( f"{detected_str} but ignoring it because one or several of --include/--exclude/--num_gpus/--num_nodes cl args were used. If you want to use CUDA_VISIBLE_DEVICES don't pass any of these arguments to deepspeed." ) @@ -333,7 +353,7 @@ def main(args=None): print(f"{detected_str}: setting --include={args.include}") del os.environ["CUDA_VISIBLE_DEVICES"] - if args.num_nodes >= 0 or args.num_gpus >= 0: + if num_nodes >= 0 or args.num_gpus >= 0: if args.include != "" or args.exclude != "": raise ValueError("Cannot specify num_nodes/gpus with include/exclude") @@ -347,7 +367,7 @@ def main(args=None): args.master_addr = "127.0.0.1" multi_node_exec = False - if not multi_node_exec and args.num_nodes > 1: + if not multi_node_exec and num_nodes > 1: raise ValueError("Num nodes is >1 but no extra nodes available via hostfile") active_resources = parse_inclusion_exclusion(resource_pool, @@ -381,10 +401,10 @@ def main(args=None): run_autotuning(args, active_resources) return - if args.num_nodes > 0: + if num_nodes > 0: updated_active_resources = collections.OrderedDict() for count, hostname in enumerate(active_resources.keys()): - if args.num_nodes == count: + if num_nodes == count: break updated_active_resources[hostname] = active_resources[hostname] active_resources = updated_active_resources @@ -418,6 +438,9 @@ def main(args=None): deepspeed_launch.append("--no_local_rank") if args.save_pid: deepspeed_launch += ["--save_pid", f"{os.getpid()}"] + if args.elastic_training: + deepspeed_launch.append("--enable_elastic_training") + deepspeed_launch.append(f"--max_nodes={args.max_num_nodes}") cmd = deepspeed_launch + [args.user_script] + args.user_args else: args.launcher = args.launcher.lower() From 9b5e72cf9890ba68be2e2a7b174e4e6f8f49b10a Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 16 Jun 2022 10:37:59 +0500 Subject: [PATCH 03/27] Remove functionAgent --- deepspeed/launcher/launch.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index b8b3ecf3010f..5aad01564a5c 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -22,8 +22,8 @@ from ..constants import TORCH_DISTRIBUTED_DEFAULT_PORT from ..utils import logger PID_FILE_BASEPATH = "/tmp" -sys.path.append('/home/t-arpanjain/work/elastic_try/using_torch/') -from functionAgent import FunctionElasticAgent +# sys.path.append('/home/t-arpanjain/work/elastic_try/using_torch/') +# from functionAgent import FunctionElasticAgent from torch.distributed.elastic.rendezvous import RendezvousParameters from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent import torch.distributed.elastic.rendezvous.registry as rdzv_registry From 4881ce8fb9ad90f14798d03d5b9a1473b55d36b7 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 16 Jun 2022 21:10:26 +0500 Subject: [PATCH 04/27] Add NCCL BLOCKING ERROR flag to elastic training --- deepspeed/launcher/launch.py | 1 + deepspeed/launcher/runner.py | 3 +++ 2 files changed, 4 insertions(+) diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 5aad01564a5c..302ea1131a13 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -221,6 +221,7 @@ def main(): os.environ["MASTER_ADDR"] = args.master_addr os.environ["MASTER_PORT"] = str(args.master_port) + os.environ["NCCL_ASYNC_ERROR_HANDLING"] = str(1) # spawn the processes cmd = [] diff --git a/deepspeed/launcher/runner.py b/deepspeed/launcher/runner.py index df55f9754618..78f0b13a6ea2 100755 --- a/deepspeed/launcher/runner.py +++ b/deepspeed/launcher/runner.py @@ -337,6 +337,9 @@ def main(args=None): num_nodes, args.max_num_nodes = parse_num_nodes(args.num_nodes, args.elastic_training) + if args.elastic_training: + assert args.master_addr != "", "Master Addr is required when elastic training is enabled" + resource_pool = fetch_hostfile(args.hostfile) # respect CUDA_VISIBLE_DEVICES for a single node and no explicit resource filters From 9f1c997b059d104429bfcda95e43b513d87dfddf Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 17 Jun 2022 03:22:37 +0500 Subject: [PATCH 05/27] transient change --- deepspeed/launcher/launch.py | 207 ++++++++++++++++++++++++++++++++++- 1 file changed, 205 insertions(+), 2 deletions(-) diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 302ea1131a13..25dee61547bd 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -30,6 +30,28 @@ from torch.distributed.elastic.agent.server.api import WorkerSpec from torch.distributed.elastic.multiprocessing import Std from typing import Any, Dict, Optional, Tuple +from torch.distributed.elastic.rendezvous.dynamic_rendezvous import ( + _BackendRendezvousStateHolder, + DynamicRendezvousHandler, + _NodeDescGenerator, + RendezvousSettings, + RendezvousTimeout, + _RendezvousState, + RendezvousBackend, + create_handler, + _get_timeout +) +from torch.distributed import Store +from torch.distributed.elastic.rendezvous import rendezvous_handler_registry +from torch.distributed.elastic.rendezvous.registry import get_rendezvous_handler +from datetime import datetime, timedelta +from torch.distributed.elastic.events import ( + NodeState, + construct_and_record_rdzv_event, +) +from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState +from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent +from torch.distributed.elastic.metrics import prof, put_metric def parse_args(): parser = ArgumentParser(description="DeepSpeed distributed training launch" @@ -126,6 +148,187 @@ def get_config_elastic(args, num_local_procs, node_rank) -> LaunchConfig: ) return config +def _remove_participant_epilogue_DS(state: _RendezvousState, settings: RendezvousSettings) -> None: + if state.complete: + # If we do not have any participants left, move to the next round. + if not state.participants: + state.complete = False + + state.round += 1 + else: + state.restart = True + else: + if len(state.participants) < settings.min_nodes: + state.deadline = None + + + +class _DSBackendRendezvousStateHolder(_BackendRendezvousStateHolder): + def _sanitize(self) -> None: + state = self._state + + if hasattr(state, 'property'): + state.restart = False + + expire_time = datetime.utcnow() - ( + self._settings.keep_alive_interval * self._settings.keep_alive_max_attempt + ) + + # Filter out the dead nodes. + self._dead_nodes = [ + node + for node, last_heartbeat in state.last_heartbeats.items() + if last_heartbeat < expire_time + ] + for node, last_heartbeat in state.last_heartbeats.items(): + print("last node:", node, last_heartbeat) + + participant_removed = False + + for dead_node in self._dead_nodes: + del state.last_heartbeats[dead_node] + + try: + del state.participants[dead_node] + + participant_removed = True + except KeyError: + pass + + try: + state.wait_list.remove(dead_node) + except KeyError: + pass + + if participant_removed: + # Common epilogue shared with the _remove_from_participants() + # function of _DistributedRendezvousOpExecutor. + _remove_participant_epilogue_DS(state, self._settings) + +class DynamicRendezvousHandlerDS(DynamicRendezvousHandler): + _node_desc_generator = _NodeDescGenerator() + + @classmethod + def from_backend( + cls, + run_id: str, + store: Store, + backend: RendezvousBackend, + min_nodes: int, + max_nodes: int, + timeout: Optional[RendezvousTimeout] = None, + ): + + node = cls._node_desc_generator.generate() + + settings = RendezvousSettings( + run_id, + min_nodes, + max_nodes, + timeout or RendezvousTimeout(), + keep_alive_interval=timedelta(seconds=5), + keep_alive_max_attempt=3, + ) + + state_holder = _DSBackendRendezvousStateHolder(backend, settings) + return cls(node, settings, backend.name, store, state_holder) + + +def create_ds_handler( + store: Store, backend: RendezvousBackend, params: RendezvousParameters +) -> DynamicRendezvousHandler: + try: + timeout = RendezvousTimeout( + _get_timeout(params, "join"), + _get_timeout(params, "last_call"), + _get_timeout(params, "close"), + ) + + return DynamicRendezvousHandlerDS.from_backend( + params.run_id, + store, + backend, + params.min_nodes, + params.max_nodes, + timeout, + ) + except Exception as e: + construct_and_record_rdzv_event( + message=f"{type(e).__name__}: {str(e)}", + run_id=params.run_id, + node_state=NodeState.FAILED, + ) + raise + +def _create_ds_c10d_handler(params: RendezvousParameters): + from torch.distributed.elastic.rendezvous.c10d_rendezvous_backend import create_backend + + backend, store = create_backend(params) + + return create_ds_handler(store, backend, params) +del rendezvous_handler_registry._registry["c10d"] +rendezvous_handler_registry.register("c10d", _create_ds_c10d_handler) + +class DSElasticAgent(LocalElasticAgent): + def _invoke_run(self, role: str = "default") -> RunResult: + # NOTE: currently only works for a single role + + spec = self._worker_group.spec + role = spec.role + + log.info( + f"[{role}] starting workers for entrypoint: {spec.get_entrypoint_name()}" + ) + + self._initialize_workers(self._worker_group) + monitor_interval = spec.monitor_interval + rdzv_handler = spec.rdzv_handler + + while True: + assert self._worker_group.state != WorkerState.INIT + time.sleep(monitor_interval) + run_result = self._monitor_workers(self._worker_group) + state = run_result.state + self._worker_group.state = state + + put_metric(f"workers.{role}.remaining_restarts", self._remaining_restarts) + put_metric(f"workers.{role}.{state.name.lower()}", 1) + + if state == WorkerState.SUCCEEDED: + log.info( + f"[{role}] worker group successfully finished." + f" Waiting {self._exit_barrier_timeout} seconds for other agents to finish." + ) + self._exit_barrier() + return run_result + elif state in {WorkerState.UNHEALTHY, WorkerState.FAILED} or rdzv_handler._state_holder.state.restart: + if self._remaining_restarts > 0: + log.info( + f"[{role}] Worker group {state.name}. " + f"{self._remaining_restarts}/{spec.max_restarts} attempts left;" + f" will restart worker group" + ) + self._remaining_restarts -= 1 + rdzv_handler._state_holder.state.restart = False + self._restart_workers(self._worker_group) + else: + self._stop_workers(self._worker_group) + self._worker_group.state = WorkerState.FAILED + self._exit_barrier() + return run_result + elif state == WorkerState.HEALTHY: + # membership changes do not count as retries + num_nodes_waiting = rdzv_handler.num_nodes_waiting() + group_rank = self._worker_group.group_rank + if num_nodes_waiting > 0: + log.info( + f"[{role}] Detected {num_nodes_waiting} " + f"new nodes from group_rank={group_rank}; " + f"will restart worker group" + ) + self._restart_workers(self._worker_group) + else: + raise Exception(f"[{role}] Worker group in {state.name} state") def main(): args = parse_args() @@ -248,7 +451,7 @@ def main(): rdzv_configs: Dict[str, str] = {'timeout': 100} rdzv_parameters = RendezvousParameters( backend='c10d', - endpoint=args.master_addr+":29401", + endpoint=args.master_addr+":29400", run_id='123456789', min_nodes=args.nnodes, max_nodes=args.max_nodes, @@ -268,7 +471,7 @@ def main(): master_addr=args.master_addr, master_port=str(args.master_port), ) - agent = LocalElasticAgent( + agent = DSElasticAgent( spec ) agent.run() From 183b6bf1f69662100e475a5774087b7973c1d23b Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 17 Jun 2022 23:22:24 +0500 Subject: [PATCH 06/27] Added DS elastic agent --- deepspeed/elasticity/__init__.py | 1 + deepspeed/elasticity/elastic_agent.py | 257 +++++++++++++++++++++++++ deepspeed/launcher/launch.py | 231 ++-------------------- deepspeed/launcher/multinode_runner.py | 1 + deepspeed/launcher/runner.py | 29 ++- 5 files changed, 297 insertions(+), 222 deletions(-) create mode 100644 deepspeed/elasticity/elastic_agent.py diff --git a/deepspeed/elasticity/__init__.py b/deepspeed/elasticity/__init__.py index be517de7df93..6635614a8dca 100644 --- a/deepspeed/elasticity/__init__.py +++ b/deepspeed/elasticity/__init__.py @@ -1 +1,2 @@ from .elasticity import compute_elastic_config, elasticity_enabled, ensure_immutable_elastic_config +from .elastic_agent import DSElasticAgent diff --git a/deepspeed/elasticity/elastic_agent.py b/deepspeed/elasticity/elastic_agent.py new file mode 100644 index 000000000000..80cf42129004 --- /dev/null +++ b/deepspeed/elasticity/elastic_agent.py @@ -0,0 +1,257 @@ +from torch.distributed.elastic.rendezvous import RendezvousParameters +from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent +import torch.distributed.elastic.rendezvous.registry as rdzv_registry +from torch.distributed.elastic.agent.server.api import WorkerSpec +from torch.distributed.elastic.multiprocessing import Std +from typing import Any, Dict, Optional, Tuple +from torch.distributed.elastic.rendezvous.dynamic_rendezvous import ( + _BackendRendezvousStateHolder, + DynamicRendezvousHandler, + _NodeDescGenerator, + RendezvousSettings, + RendezvousTimeout, + _RendezvousState, + RendezvousBackend, + create_handler, + _get_timeout, + _RendezvousState +) +from torch.distributed import Store +from torch.distributed.elastic.rendezvous import rendezvous_handler_registry +from torch.distributed.elastic.rendezvous.registry import get_rendezvous_handler +from datetime import datetime, timedelta +from torch.distributed.elastic.events import ( + NodeState, + construct_and_record_rdzv_event, +) +from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState +from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent +from torch.distributed.elastic.metrics import prof, put_metric +import time + + +def _remove_participant_epilogue_DS(state: _RendezvousState, settings: RendezvousSettings) -> None: + if state.complete: + # If we do not have any participants left, move to the next round. + if not state.participants: + state.complete = False + + state.round += 1 + else: + state.complete = False + + state.round += 1 + else: + if len(state.participants) < settings.min_nodes: + state.deadline = None + + +# class _RendezvousStateDS(_RendezvousState): +# round: int +# complete: bool +# deadline: Optional[datetime] +# closed: bool +# participants: Dict[_NodeDesc, int] +# wait_list: Set[_NodeDesc] +# last_heartbeats: Dict[_NodeDesc, datetime] +# def __init__(self): +# super().__init__() +# self.restart = False + + +class _DSBackendRendezvousStateHolder(_BackendRendezvousStateHolder): + # def __init__( + # self, + # backend: RendezvousBackend, + # settings: RendezvousSettings, + # cache_duration: int = 1, + # ) -> None: + # super().__init__(backend, settings, cache_duration) + # self._state = _RendezvousStateDS() + # print(self._state.restart) + + # def state(self) -> _RendezvousStateDS: + # """See base class.""" + # return self._state + def _sanitize(self) -> None: + state = self._state + + if not hasattr(state, 'property'): + state.restart = False + + expire_time = datetime.utcnow() - ( + self._settings.keep_alive_interval * self._settings.keep_alive_max_attempt + ) + + # Filter out the dead nodes. + self._dead_nodes = [ + node + for node, last_heartbeat in state.last_heartbeats.items() + if last_heartbeat < expire_time + ] + # for node, last_heartbeat in state.last_heartbeats.items(): + # print("last node:", node, last_heartbeat) + + participant_removed = False + + for dead_node in self._dead_nodes: + del state.last_heartbeats[dead_node] + + try: + del state.participants[dead_node] + + participant_removed = True + except KeyError: + pass + + try: + state.wait_list.remove(dead_node) + except KeyError: + pass + + if participant_removed: + # Common epilogue shared with the _remove_from_participants() + # function of _DistributedRendezvousOpExecutor. + _remove_participant_epilogue_DS(state, self._settings) + +class DynamicRendezvousHandlerDS(DynamicRendezvousHandler): + _node_desc_generator = _NodeDescGenerator() + + @classmethod + def from_backend( + cls, + run_id: str, + store: Store, + backend: RendezvousBackend, + min_nodes: int, + max_nodes: int, + timeout: Optional[RendezvousTimeout] = None, + ): + + node = cls._node_desc_generator.generate() + + settings = RendezvousSettings( + run_id, + min_nodes, + max_nodes, + timeout or RendezvousTimeout(), + keep_alive_interval=timedelta(seconds=5), + keep_alive_max_attempt=3, + ) + + state_holder = _DSBackendRendezvousStateHolder(backend, settings) + return cls(node, settings, backend.name, store, state_holder) + + +def create_ds_handler( + store: Store, backend: RendezvousBackend, params: RendezvousParameters +) -> DynamicRendezvousHandler: + try: + timeout = RendezvousTimeout( + _get_timeout(params, "join"), + _get_timeout(params, "last_call"), + _get_timeout(params, "close"), + ) + + return DynamicRendezvousHandlerDS.from_backend( + params.run_id, + store, + backend, + params.min_nodes, + params.max_nodes, + timeout, + ) + except Exception as e: + construct_and_record_rdzv_event( + message=f"{type(e).__name__}: {str(e)}", + run_id=params.run_id, + node_state=NodeState.FAILED, + ) + raise + +def _create_ds_c10d_handler(params: RendezvousParameters): + from torch.distributed.elastic.rendezvous.c10d_rendezvous_backend import create_backend + + backend, store = create_backend(params) + + return create_ds_handler(store, backend, params) +# del rendezvous_handler_registry._registry["c10d"] +# rendezvous_handler_registry.register("c10d", _create_ds_c10d_handler) + +class DSElasticAgent(LocalElasticAgent): + def _invoke_run(self, role: str = "default") -> RunResult: + # NOTE: currently only works for a single role + + spec = self._worker_group.spec + role = spec.role + + log.info( + f"[{role}] starting workers for entrypoint: {spec.get_entrypoint_name()}" + ) + + self._initialize_workers(self._worker_group) + monitor_interval = spec.monitor_interval + rdzv_handler = spec.rdzv_handler + + # if not hasattr(rdzv_handler._state_holder.state, 'property'): + # rdzv_handler._state_holder.state.restart = False + + participants = rdzv_handler._state_holder.state.participants + + while True: + assert self._worker_group.state != WorkerState.INIT + time.sleep(monitor_interval) + run_result = self._monitor_workers(self._worker_group) + state = run_result.state + self._worker_group.state = state + + expire_time = datetime.utcnow() - ( + rdzv_handler._settings.keep_alive_interval * rdzv_handler._settings.keep_alive_max_attempt + ) + _dead_nodes = [ + node + for node, last_heartbeat in rdzv_handler._state_holder.state.last_heartbeats.items() + if last_heartbeat < expire_time + ] + + put_metric(f"workers.{role}.remaining_restarts", self._remaining_restarts) + put_metric(f"workers.{role}.{state.name.lower()}", 1) + + if state == WorkerState.SUCCEEDED: + log.info( + f"[{role}] worker group successfully finished." + f" Waiting {self._exit_barrier_timeout} seconds for other agents to finish." + ) + self._exit_barrier() + return run_result + elif state in {WorkerState.UNHEALTHY, WorkerState.FAILED} or len(participants)>len( rdzv_handler._state_holder.state.participants): + if self._remaining_restarts > 0: + log.info( + f"[{role}] Worker group {state.name}. " + f"{self._remaining_restarts}/{spec.max_restarts} attempts left;" + f" will restart worker group" + ) + self._remaining_restarts -= 1 + # rdzv_handler._state_holder.state.restart = False + self._restart_workers(self._worker_group) + participants = rdzv_handler._state_holder.state.participants + + else: + self._stop_workers(self._worker_group) + self._worker_group.state = WorkerState.FAILED + self._exit_barrier() + return run_result + elif state == WorkerState.HEALTHY: + # membership changes do not count as retries + num_nodes_waiting = rdzv_handler.num_nodes_waiting() + group_rank = self._worker_group.group_rank + if num_nodes_waiting > 0: + log.info( + f"[{role}] Detected {num_nodes_waiting} " + f"new nodes from group_rank={group_rank}; " + f"will restart worker group" + ) + self._restart_workers(self._worker_group) + participants = rdzv_handler._state_holder.state.participants + else: + raise Exception(f"[{role}] Worker group in {state.name} state") diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 25dee61547bd..f32a0f2bf99c 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -21,37 +21,13 @@ from torch.distributed.elastic.multiprocessing import Std from ..constants import TORCH_DISTRIBUTED_DEFAULT_PORT from ..utils import logger -PID_FILE_BASEPATH = "/tmp" -# sys.path.append('/home/t-arpanjain/work/elastic_try/using_torch/') -# from functionAgent import FunctionElasticAgent +from ..elasticity import DSElasticAgent from torch.distributed.elastic.rendezvous import RendezvousParameters -from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent -import torch.distributed.elastic.rendezvous.registry as rdzv_registry from torch.distributed.elastic.agent.server.api import WorkerSpec -from torch.distributed.elastic.multiprocessing import Std -from typing import Any, Dict, Optional, Tuple -from torch.distributed.elastic.rendezvous.dynamic_rendezvous import ( - _BackendRendezvousStateHolder, - DynamicRendezvousHandler, - _NodeDescGenerator, - RendezvousSettings, - RendezvousTimeout, - _RendezvousState, - RendezvousBackend, - create_handler, - _get_timeout -) -from torch.distributed import Store -from torch.distributed.elastic.rendezvous import rendezvous_handler_registry -from torch.distributed.elastic.rendezvous.registry import get_rendezvous_handler -from datetime import datetime, timedelta -from torch.distributed.elastic.events import ( - NodeState, - construct_and_record_rdzv_event, -) -from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState -from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent -from torch.distributed.elastic.metrics import prof, put_metric +import torch.distributed.elastic.rendezvous.registry as rdzv_registry +PID_FILE_BASEPATH = "/tmp" + + def parse_args(): parser = ArgumentParser(description="DeepSpeed distributed training launch" @@ -97,6 +73,11 @@ def parse_args(): action="store_true", help="Enable elastic training support.") + parser.add_argument("--min_nodes", + type=int, + default=-1, + help="Min number of nodes in elastic training.") + parser.add_argument("--max_nodes", type=int, default=-1, @@ -148,188 +129,6 @@ def get_config_elastic(args, num_local_procs, node_rank) -> LaunchConfig: ) return config -def _remove_participant_epilogue_DS(state: _RendezvousState, settings: RendezvousSettings) -> None: - if state.complete: - # If we do not have any participants left, move to the next round. - if not state.participants: - state.complete = False - - state.round += 1 - else: - state.restart = True - else: - if len(state.participants) < settings.min_nodes: - state.deadline = None - - - -class _DSBackendRendezvousStateHolder(_BackendRendezvousStateHolder): - def _sanitize(self) -> None: - state = self._state - - if hasattr(state, 'property'): - state.restart = False - - expire_time = datetime.utcnow() - ( - self._settings.keep_alive_interval * self._settings.keep_alive_max_attempt - ) - - # Filter out the dead nodes. - self._dead_nodes = [ - node - for node, last_heartbeat in state.last_heartbeats.items() - if last_heartbeat < expire_time - ] - for node, last_heartbeat in state.last_heartbeats.items(): - print("last node:", node, last_heartbeat) - - participant_removed = False - - for dead_node in self._dead_nodes: - del state.last_heartbeats[dead_node] - - try: - del state.participants[dead_node] - - participant_removed = True - except KeyError: - pass - - try: - state.wait_list.remove(dead_node) - except KeyError: - pass - - if participant_removed: - # Common epilogue shared with the _remove_from_participants() - # function of _DistributedRendezvousOpExecutor. - _remove_participant_epilogue_DS(state, self._settings) - -class DynamicRendezvousHandlerDS(DynamicRendezvousHandler): - _node_desc_generator = _NodeDescGenerator() - - @classmethod - def from_backend( - cls, - run_id: str, - store: Store, - backend: RendezvousBackend, - min_nodes: int, - max_nodes: int, - timeout: Optional[RendezvousTimeout] = None, - ): - - node = cls._node_desc_generator.generate() - - settings = RendezvousSettings( - run_id, - min_nodes, - max_nodes, - timeout or RendezvousTimeout(), - keep_alive_interval=timedelta(seconds=5), - keep_alive_max_attempt=3, - ) - - state_holder = _DSBackendRendezvousStateHolder(backend, settings) - return cls(node, settings, backend.name, store, state_holder) - - -def create_ds_handler( - store: Store, backend: RendezvousBackend, params: RendezvousParameters -) -> DynamicRendezvousHandler: - try: - timeout = RendezvousTimeout( - _get_timeout(params, "join"), - _get_timeout(params, "last_call"), - _get_timeout(params, "close"), - ) - - return DynamicRendezvousHandlerDS.from_backend( - params.run_id, - store, - backend, - params.min_nodes, - params.max_nodes, - timeout, - ) - except Exception as e: - construct_and_record_rdzv_event( - message=f"{type(e).__name__}: {str(e)}", - run_id=params.run_id, - node_state=NodeState.FAILED, - ) - raise - -def _create_ds_c10d_handler(params: RendezvousParameters): - from torch.distributed.elastic.rendezvous.c10d_rendezvous_backend import create_backend - - backend, store = create_backend(params) - - return create_ds_handler(store, backend, params) -del rendezvous_handler_registry._registry["c10d"] -rendezvous_handler_registry.register("c10d", _create_ds_c10d_handler) - -class DSElasticAgent(LocalElasticAgent): - def _invoke_run(self, role: str = "default") -> RunResult: - # NOTE: currently only works for a single role - - spec = self._worker_group.spec - role = spec.role - - log.info( - f"[{role}] starting workers for entrypoint: {spec.get_entrypoint_name()}" - ) - - self._initialize_workers(self._worker_group) - monitor_interval = spec.monitor_interval - rdzv_handler = spec.rdzv_handler - - while True: - assert self._worker_group.state != WorkerState.INIT - time.sleep(monitor_interval) - run_result = self._monitor_workers(self._worker_group) - state = run_result.state - self._worker_group.state = state - - put_metric(f"workers.{role}.remaining_restarts", self._remaining_restarts) - put_metric(f"workers.{role}.{state.name.lower()}", 1) - - if state == WorkerState.SUCCEEDED: - log.info( - f"[{role}] worker group successfully finished." - f" Waiting {self._exit_barrier_timeout} seconds for other agents to finish." - ) - self._exit_barrier() - return run_result - elif state in {WorkerState.UNHEALTHY, WorkerState.FAILED} or rdzv_handler._state_holder.state.restart: - if self._remaining_restarts > 0: - log.info( - f"[{role}] Worker group {state.name}. " - f"{self._remaining_restarts}/{spec.max_restarts} attempts left;" - f" will restart worker group" - ) - self._remaining_restarts -= 1 - rdzv_handler._state_holder.state.restart = False - self._restart_workers(self._worker_group) - else: - self._stop_workers(self._worker_group) - self._worker_group.state = WorkerState.FAILED - self._exit_barrier() - return run_result - elif state == WorkerState.HEALTHY: - # membership changes do not count as retries - num_nodes_waiting = rdzv_handler.num_nodes_waiting() - group_rank = self._worker_group.group_rank - if num_nodes_waiting > 0: - log.info( - f"[{role}] Detected {num_nodes_waiting} " - f"new nodes from group_rank={group_rank}; " - f"will restart worker group" - ) - self._restart_workers(self._worker_group) - else: - raise Exception(f"[{role}] Worker group in {state.name} state") - def main(): args = parse_args() current_env = os.environ.copy() @@ -420,12 +219,18 @@ def main(): # dist_rank = global_rank_mapping[local_node][local_rank] # os.environ["RANK"] = str(dist_rank) # os.environ["LOCAL_RANK"] = str(local_rank) - assert args.max_nodes > 0, "Max nodes should be provided in elastic training" + if args.min_nodes== -1: + args.min_nodes = 1 + if args.max_nodes== -1: + args.max_nodes = args.nnodes + assert args.max_nodes > 0 and args.min_nodes > 0 , "Max and Min nodes should be positive" os.environ["MASTER_ADDR"] = args.master_addr os.environ["MASTER_PORT"] = str(args.master_port) os.environ["NCCL_ASYNC_ERROR_HANDLING"] = str(1) + + # spawn the processes cmd = [] if not args.no_python: @@ -453,7 +258,7 @@ def main(): backend='c10d', endpoint=args.master_addr+":29400", run_id='123456789', - min_nodes=args.nnodes, + min_nodes=args.min_nodes, max_nodes=args.max_nodes, **rdzv_configs ) diff --git a/deepspeed/launcher/multinode_runner.py b/deepspeed/launcher/multinode_runner.py index 2e885bc47c6f..5b9919483310 100644 --- a/deepspeed/launcher/multinode_runner.py +++ b/deepspeed/launcher/multinode_runner.py @@ -97,6 +97,7 @@ def get_cmd(self, environment, active_resources): if self.args.elastic_training: deepspeed_launch.append("--enable_elastic_training") deepspeed_launch.append(f"--max_nodes={self.args.max_num_nodes}") + deepspeed_launch.append(f"--min_nodes={self.args.min_num_nodes}") return pdsh_cmd_args + deepspeed_launch + [self.user_script ] + self.user_arguments diff --git a/deepspeed/launcher/runner.py b/deepspeed/launcher/runner.py index 78f0b13a6ea2..58b38d22017a 100755 --- a/deepspeed/launcher/runner.py +++ b/deepspeed/launcher/runner.py @@ -70,11 +70,23 @@ def parse_args(args=None): ''') parser.add_argument("--num_nodes", - type=str, - default="-1", + type=int, + default=-1, help="Total number of worker nodes to run on, this will use " "the top N hosts from the given hostfile.") + parser.add_argument("--min_num_nodes", + type=int, + default=-1, + help="Minimum number of nodes to run elastic training on. " + "Default is 1 when elastic training is enabled") + + parser.add_argument("--max_num_nodes", + type=int, + default=-1, + help="Maximum number of nodes to run elastic training on. " + "Default is num_nodes when elastic training is enabled") + parser.add_argument("--num_gpus", type=int, default=-1, @@ -335,8 +347,6 @@ def parse_num_nodes(str_num_nodes: str, elastic_training: bool): def main(args=None): args = parse_args(args) - num_nodes, args.max_num_nodes = parse_num_nodes(args.num_nodes, args.elastic_training) - if args.elastic_training: assert args.master_addr != "", "Master Addr is required when elastic training is enabled" @@ -347,7 +357,7 @@ def main(args=None): if not resource_pool and len(cuda_visible_devices): detected_str = f"Detected CUDA_VISIBLE_DEVICES={cuda_visible_devices}" if len(args.include) or len( - args.exclude) or num_nodes > 1 or args.num_gpus > 0: + args.exclude) or args.num_nodes > 1 or args.num_gpus > 0: print( f"{detected_str} but ignoring it because one or several of --include/--exclude/--num_gpus/--num_nodes cl args were used. If you want to use CUDA_VISIBLE_DEVICES don't pass any of these arguments to deepspeed." ) @@ -356,7 +366,7 @@ def main(args=None): print(f"{detected_str}: setting --include={args.include}") del os.environ["CUDA_VISIBLE_DEVICES"] - if num_nodes >= 0 or args.num_gpus >= 0: + if args.num_nodes >= 0 or args.num_gpus >= 0: if args.include != "" or args.exclude != "": raise ValueError("Cannot specify num_nodes/gpus with include/exclude") @@ -370,7 +380,7 @@ def main(args=None): args.master_addr = "127.0.0.1" multi_node_exec = False - if not multi_node_exec and num_nodes > 1: + if not multi_node_exec and args.num_nodes > 1: raise ValueError("Num nodes is >1 but no extra nodes available via hostfile") active_resources = parse_inclusion_exclusion(resource_pool, @@ -404,10 +414,10 @@ def main(args=None): run_autotuning(args, active_resources) return - if num_nodes > 0: + if args.num_nodes > 0: updated_active_resources = collections.OrderedDict() for count, hostname in enumerate(active_resources.keys()): - if num_nodes == count: + if args.num_nodes == count: break updated_active_resources[hostname] = active_resources[hostname] active_resources = updated_active_resources @@ -444,6 +454,7 @@ def main(args=None): if args.elastic_training: deepspeed_launch.append("--enable_elastic_training") deepspeed_launch.append(f"--max_nodes={args.max_num_nodes}") + deepspeed_launch.append(f"--min_nodes={args.min_num_nodes}") cmd = deepspeed_launch + [args.user_script] + args.user_args else: args.launcher = args.launcher.lower() From 4d024f07e5225356985f2bfb1e157659626b3423 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 1 Jul 2022 02:37:09 +0500 Subject: [PATCH 07/27] Cleanup --- deepspeed/constants.py | 3 +- deepspeed/elasticity/elastic_agent.py | 175 -------------------------- deepspeed/launcher/launch.py | 16 +-- 3 files changed, 3 insertions(+), 191 deletions(-) diff --git a/deepspeed/constants.py b/deepspeed/constants.py index 318d2461bfe4..9e7f511e443f 100644 --- a/deepspeed/constants.py +++ b/deepspeed/constants.py @@ -13,7 +13,6 @@ # (only if NCCL_BLOCKING_WAIT or NCCL_ASYNC_ERROR_HANDLING is set to 1). # To make an attempt at backwards compatibility with THD, we use an # extraordinarily high default timeout, given that THD did not have timeouts. -# default_pg_timeout = timedelta(minutes=30) -default_pg_timeout = timedelta(seconds=30) +default_pg_timeout = timedelta(minutes=30) INFERENCE_GENERIC_MODE = 'generic' INFERENCE_SPECIALIZED_MODE = 'specialized' diff --git a/deepspeed/elasticity/elastic_agent.py b/deepspeed/elasticity/elastic_agent.py index 80cf42129004..44c75cc59e04 100644 --- a/deepspeed/elasticity/elastic_agent.py +++ b/deepspeed/elasticity/elastic_agent.py @@ -1,183 +1,11 @@ -from torch.distributed.elastic.rendezvous import RendezvousParameters from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent -import torch.distributed.elastic.rendezvous.registry as rdzv_registry -from torch.distributed.elastic.agent.server.api import WorkerSpec -from torch.distributed.elastic.multiprocessing import Std from typing import Any, Dict, Optional, Tuple -from torch.distributed.elastic.rendezvous.dynamic_rendezvous import ( - _BackendRendezvousStateHolder, - DynamicRendezvousHandler, - _NodeDescGenerator, - RendezvousSettings, - RendezvousTimeout, - _RendezvousState, - RendezvousBackend, - create_handler, - _get_timeout, - _RendezvousState -) -from torch.distributed import Store -from torch.distributed.elastic.rendezvous import rendezvous_handler_registry -from torch.distributed.elastic.rendezvous.registry import get_rendezvous_handler from datetime import datetime, timedelta -from torch.distributed.elastic.events import ( - NodeState, - construct_and_record_rdzv_event, -) from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState -from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent from torch.distributed.elastic.metrics import prof, put_metric import time -def _remove_participant_epilogue_DS(state: _RendezvousState, settings: RendezvousSettings) -> None: - if state.complete: - # If we do not have any participants left, move to the next round. - if not state.participants: - state.complete = False - - state.round += 1 - else: - state.complete = False - - state.round += 1 - else: - if len(state.participants) < settings.min_nodes: - state.deadline = None - - -# class _RendezvousStateDS(_RendezvousState): -# round: int -# complete: bool -# deadline: Optional[datetime] -# closed: bool -# participants: Dict[_NodeDesc, int] -# wait_list: Set[_NodeDesc] -# last_heartbeats: Dict[_NodeDesc, datetime] -# def __init__(self): -# super().__init__() -# self.restart = False - - -class _DSBackendRendezvousStateHolder(_BackendRendezvousStateHolder): - # def __init__( - # self, - # backend: RendezvousBackend, - # settings: RendezvousSettings, - # cache_duration: int = 1, - # ) -> None: - # super().__init__(backend, settings, cache_duration) - # self._state = _RendezvousStateDS() - # print(self._state.restart) - - # def state(self) -> _RendezvousStateDS: - # """See base class.""" - # return self._state - def _sanitize(self) -> None: - state = self._state - - if not hasattr(state, 'property'): - state.restart = False - - expire_time = datetime.utcnow() - ( - self._settings.keep_alive_interval * self._settings.keep_alive_max_attempt - ) - - # Filter out the dead nodes. - self._dead_nodes = [ - node - for node, last_heartbeat in state.last_heartbeats.items() - if last_heartbeat < expire_time - ] - # for node, last_heartbeat in state.last_heartbeats.items(): - # print("last node:", node, last_heartbeat) - - participant_removed = False - - for dead_node in self._dead_nodes: - del state.last_heartbeats[dead_node] - - try: - del state.participants[dead_node] - - participant_removed = True - except KeyError: - pass - - try: - state.wait_list.remove(dead_node) - except KeyError: - pass - - if participant_removed: - # Common epilogue shared with the _remove_from_participants() - # function of _DistributedRendezvousOpExecutor. - _remove_participant_epilogue_DS(state, self._settings) - -class DynamicRendezvousHandlerDS(DynamicRendezvousHandler): - _node_desc_generator = _NodeDescGenerator() - - @classmethod - def from_backend( - cls, - run_id: str, - store: Store, - backend: RendezvousBackend, - min_nodes: int, - max_nodes: int, - timeout: Optional[RendezvousTimeout] = None, - ): - - node = cls._node_desc_generator.generate() - - settings = RendezvousSettings( - run_id, - min_nodes, - max_nodes, - timeout or RendezvousTimeout(), - keep_alive_interval=timedelta(seconds=5), - keep_alive_max_attempt=3, - ) - - state_holder = _DSBackendRendezvousStateHolder(backend, settings) - return cls(node, settings, backend.name, store, state_holder) - - -def create_ds_handler( - store: Store, backend: RendezvousBackend, params: RendezvousParameters -) -> DynamicRendezvousHandler: - try: - timeout = RendezvousTimeout( - _get_timeout(params, "join"), - _get_timeout(params, "last_call"), - _get_timeout(params, "close"), - ) - - return DynamicRendezvousHandlerDS.from_backend( - params.run_id, - store, - backend, - params.min_nodes, - params.max_nodes, - timeout, - ) - except Exception as e: - construct_and_record_rdzv_event( - message=f"{type(e).__name__}: {str(e)}", - run_id=params.run_id, - node_state=NodeState.FAILED, - ) - raise - -def _create_ds_c10d_handler(params: RendezvousParameters): - from torch.distributed.elastic.rendezvous.c10d_rendezvous_backend import create_backend - - backend, store = create_backend(params) - - return create_ds_handler(store, backend, params) -# del rendezvous_handler_registry._registry["c10d"] -# rendezvous_handler_registry.register("c10d", _create_ds_c10d_handler) - class DSElasticAgent(LocalElasticAgent): def _invoke_run(self, role: str = "default") -> RunResult: # NOTE: currently only works for a single role @@ -193,9 +21,6 @@ def _invoke_run(self, role: str = "default") -> RunResult: monitor_interval = spec.monitor_interval rdzv_handler = spec.rdzv_handler - # if not hasattr(rdzv_handler._state_holder.state, 'property'): - # rdzv_handler._state_holder.state.restart = False - participants = rdzv_handler._state_holder.state.participants while True: diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index f32a0f2bf99c..adf4aa38a19f 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -216,9 +216,6 @@ def main(): process = subprocess.Popen(cmd, env=current_env) processes.append(process) else: - # dist_rank = global_rank_mapping[local_node][local_rank] - # os.environ["RANK"] = str(dist_rank) - # os.environ["LOCAL_RANK"] = str(local_rank) if args.min_nodes== -1: args.min_nodes = 1 if args.max_nodes== -1: @@ -229,9 +226,7 @@ def main(): os.environ["MASTER_PORT"] = str(args.master_port) os.environ["NCCL_ASYNC_ERROR_HANDLING"] = str(1) - - - # spawn the processes + # Get config and arguments cmd = [] if not args.no_python: cmd = [sys.executable, "-u"] @@ -245,14 +240,7 @@ def main(): cmd += args.training_script_args elastic_config = get_config_elastic(args, num_local_procs,args.node_rank) cmd_args = cmd[1:] - # cmd_args = ['MASTER_ADDR={}'.format(args.master_addr), 'MASTER_PORT={}'.format(args.master_port)] + cmd_args - print ("CMD is:",cmd_args) - - - # elastic_launch( - # config=elastic_config, - # entrypoint= cmd[0], - # )(*cmd_args) + rdzv_configs: Dict[str, str] = {'timeout': 100} rdzv_parameters = RendezvousParameters( backend='c10d', From 1ab31049d5800605c9102cb3c9b854fd9a94d23d Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 1 Jul 2022 21:58:02 +0500 Subject: [PATCH 08/27] pass environment variables to worker processes --- deepspeed/elasticity/elastic_agent.py | 89 +++++++++++++++++++++++++++ deepspeed/launcher/launch.py | 9 ++- 2 files changed, 93 insertions(+), 5 deletions(-) diff --git a/deepspeed/elasticity/elastic_agent.py b/deepspeed/elasticity/elastic_agent.py index 44c75cc59e04..c0606b680464 100644 --- a/deepspeed/elasticity/elastic_agent.py +++ b/deepspeed/elasticity/elastic_agent.py @@ -3,10 +3,99 @@ from datetime import datetime, timedelta from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState from torch.distributed.elastic.metrics import prof, put_metric +from torch.distributed.elastic.agent.server.api import ( + RunResult, + WorkerGroup, + WorkerSpec, + WorkerState, +) import time +import os +from torch.distributed.elastic.multiprocessing import PContext, start_processes +from torch.distributed.elastic.utils import macros +import shutil +import copy class DSElasticAgent(LocalElasticAgent): + + def __init__( + self, + spec: WorkerSpec, + env: Dict, + start_method="spawn", + exit_barrier_timeout: float = 300, + log_dir: Optional[str] = None, + ): + super().__init__(spec, start_method, exit_barrier_timeout, log_dir) + self.ds_env = env + + def _start_workers(self, worker_group: WorkerGroup) -> Dict[int, Any]: + spec = worker_group.spec + store = worker_group.store + assert store is not None + master_addr, master_port = super()._get_master_addr_port(store) + restart_count = spec.max_restarts - self._remaining_restarts + + use_agent_store = spec.rdzv_handler.get_backend() == "static" + + args: Dict[int, Tuple] = {} + envs: Dict[int, Dict[str, str]] = {} + for worker in worker_group.workers: + local_rank = worker.local_rank + + worker_env_ds = copy.deepcopy(self.ds_env) + worker_env_elastic = { + "LOCAL_RANK": str(local_rank), + "RANK": str(worker.global_rank), + "GROUP_RANK": str(worker_group.group_rank), + "ROLE_RANK": str(worker.role_rank), + "ROLE_NAME": spec.role, + "LOCAL_WORLD_SIZE": str(spec.local_world_size), + "WORLD_SIZE": str(worker.world_size), + "GROUP_WORLD_SIZE": str(worker_group.group_world_size), + "ROLE_WORLD_SIZE": str(worker.role_world_size), + "MASTER_ADDR": master_addr, + "MASTER_PORT": str(master_port), + "TORCHELASTIC_RESTART_COUNT": str(restart_count), + "TORCHELASTIC_MAX_RESTARTS": str(spec.max_restarts), + "TORCHELASTIC_RUN_ID": spec.rdzv_handler.get_run_id(), + "TORCHELASTIC_USE_AGENT_STORE": str(use_agent_store), + "NCCL_ASYNC_ERROR_HANDLING": os.getenv( + "NCCL_ASYNC_ERROR_HANDLING", str(1) + ), + } + worker_env_ds.update(worker_env_elastic) + if "OMP_NUM_THREADS" in os.environ: + worker_env_ds["OMP_NUM_THREADS"] = os.environ["OMP_NUM_THREADS"] + + envs[local_rank] = worker_env_ds + worker_args = list(spec.args) + worker_args = macros.substitute(worker_args, str(local_rank)) + args[local_rank] = tuple(worker_args) + + # scaling events do not count towards restarts (gets same attempt #) + # remove existing log dir if this restart is due to a scaling event + attempt_log_dir = os.path.join(self._log_dir, f"attempt_{restart_count}") + shutil.rmtree(attempt_log_dir, ignore_errors=True) + os.makedirs(attempt_log_dir) + + assert spec.entrypoint is not None + self._pcontext = start_processes( + name=spec.role, + entrypoint=spec.entrypoint, + args=args, + envs=envs, + log_dir=attempt_log_dir, + start_method=self._start_method, + redirects=spec.redirects, + tee=spec.tee, + ) + + return self._pcontext.pids() + + + def _invoke_run(self, role: str = "default") -> RunResult: # NOTE: currently only works for a single role diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index adf4aa38a19f..63b2fd456847 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -221,10 +221,8 @@ def main(): if args.max_nodes== -1: args.max_nodes = args.nnodes assert args.max_nodes > 0 and args.min_nodes > 0 , "Max and Min nodes should be positive" - - os.environ["MASTER_ADDR"] = args.master_addr - os.environ["MASTER_PORT"] = str(args.master_port) - os.environ["NCCL_ASYNC_ERROR_HANDLING"] = str(1) + + current_env["NCCL_ASYNC_ERROR_HANDLING"] = str(1) # Get config and arguments cmd = [] @@ -265,7 +263,8 @@ def main(): master_port=str(args.master_port), ) agent = DSElasticAgent( - spec + spec, + current_env ) agent.run() From f55b767cb5a31b2f58fe2251cb837feef738a96b Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 1 Jul 2022 23:23:26 +0500 Subject: [PATCH 09/27] Enable elastic checkpoint for scale down in elastic training --- deepspeed/runtime/engine.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/deepspeed/runtime/engine.py b/deepspeed/runtime/engine.py index 731708549ab6..dd5c7f6000db 100644 --- a/deepspeed/runtime/engine.py +++ b/deepspeed/runtime/engine.py @@ -2677,7 +2677,8 @@ def _load_zero_checkpoint(self, load_dir, tag, load_optimizer_states=True): if zero_sd_list is None: return False - if load_optimizer_states and self.dp_world_size != self.loaded_checkpoint_dp_world_size: + if load_optimizer_states and self.dp_world_size != self.loaded_checkpoint_dp_world_size \ + and not self.zero_elastic_checkpoint(): raise ZeRORuntimeException("The checkpoint being loaded used a DP " \ f"world size of {self.loaded_checkpoint_dp_world_size} but the " \ f"current world size is {self.dp_world_size}. Automatic adjustment " \ From e37c7619ccf6643b12b6795c7055bfe72e469dec Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Sat, 2 Jul 2022 11:33:42 +0500 Subject: [PATCH 10/27] added detection of master addr and port on rank 0 --- deepspeed/elasticity/elastic_agent.py | 23 +++++++++++++++++++++-- deepspeed/launcher/launch.py | 6 +++--- 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/deepspeed/elasticity/elastic_agent.py b/deepspeed/elasticity/elastic_agent.py index c0606b680464..f329197643ea 100644 --- a/deepspeed/elasticity/elastic_agent.py +++ b/deepspeed/elasticity/elastic_agent.py @@ -1,7 +1,7 @@ from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent from typing import Any, Dict, Optional, Tuple from datetime import datetime, timedelta -from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState +from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState, _get_socket_with_port from torch.distributed.elastic.metrics import prof, put_metric from torch.distributed.elastic.agent.server.api import ( RunResult, @@ -9,13 +9,15 @@ WorkerSpec, WorkerState, ) +from torch.distributed import Store import time import os from torch.distributed.elastic.multiprocessing import PContext, start_processes from torch.distributed.elastic.utils import macros import shutil import copy - +from contextlib import closing +import subprocess class DSElasticAgent(LocalElasticAgent): @@ -29,6 +31,23 @@ def __init__( ): super().__init__(spec, start_method, exit_barrier_timeout, log_dir) self.ds_env = env + + @staticmethod + def _set_master_addr_port( + store: Store, master_addr: Optional[str], master_port: Optional[int] + ): + if master_port is None: + sock = _get_socket_with_port() + with closing(sock): + master_port = sock.getsockname()[1] + + if master_addr is None: + # master_addr = _get_fq_hostname() + result = subprocess.check_output("hostname -I", shell=True) + master_addr = result.decode('utf-8').split()[0] + + store.set("MASTER_ADDR", master_addr.encode(encoding="UTF-8")) + store.set("MASTER_PORT", str(master_port).encode(encoding="UTF-8")) def _start_workers(self, worker_group: WorkerGroup) -> Dict[int, Any]: spec = worker_group.spec diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 63b2fd456847..23ad54bf3194 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -242,7 +242,7 @@ def main(): rdzv_configs: Dict[str, str] = {'timeout': 100} rdzv_parameters = RendezvousParameters( backend='c10d', - endpoint=args.master_addr+":29400", + endpoint=args.master_addr+":"+str(args.master_port), run_id='123456789', min_nodes=args.min_nodes, max_nodes=args.max_nodes, @@ -259,8 +259,8 @@ def main(): monitor_interval=5, redirects=Std.from_str("0"), tee=Std.from_str("0"), - master_addr=args.master_addr, - master_port=str(args.master_port), + master_addr=None, + master_port=None, ) agent = DSElasticAgent( spec, From a41757d042b0c48d39d9d4be13a565c4fb6e3118 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Wed, 6 Jul 2022 02:05:19 +0500 Subject: [PATCH 11/27] fixed formatting --- deepspeed/elasticity/elastic_agent.py | 53 +++++++++++------------ deepspeed/launcher/launch.py | 62 ++++++++++++--------------- deepspeed/launcher/runner.py | 5 ++- 3 files changed, 56 insertions(+), 64 deletions(-) diff --git a/deepspeed/elasticity/elastic_agent.py b/deepspeed/elasticity/elastic_agent.py index f329197643ea..e9ac126b9a84 100644 --- a/deepspeed/elasticity/elastic_agent.py +++ b/deepspeed/elasticity/elastic_agent.py @@ -9,7 +9,7 @@ WorkerSpec, WorkerState, ) -from torch.distributed import Store +from torch.distributed import Store import time import os from torch.distributed.elastic.multiprocessing import PContext, start_processes @@ -19,8 +19,8 @@ from contextlib import closing import subprocess -class DSElasticAgent(LocalElasticAgent): +class DSElasticAgent(LocalElasticAgent): def __init__( self, spec: WorkerSpec, @@ -33,9 +33,9 @@ def __init__( self.ds_env = env @staticmethod - def _set_master_addr_port( - store: Store, master_addr: Optional[str], master_port: Optional[int] - ): + def _set_master_addr_port(store: Store, + master_addr: Optional[str], + master_port: Optional[int]): if master_port is None: sock = _get_socket_with_port() with closing(sock): @@ -48,7 +48,7 @@ def _set_master_addr_port( store.set("MASTER_ADDR", master_addr.encode(encoding="UTF-8")) store.set("MASTER_PORT", str(master_port).encode(encoding="UTF-8")) - + def _start_workers(self, worker_group: WorkerGroup) -> Dict[int, Any]: spec = worker_group.spec store = worker_group.store @@ -80,9 +80,8 @@ def _start_workers(self, worker_group: WorkerGroup) -> Dict[int, Any]: "TORCHELASTIC_MAX_RESTARTS": str(spec.max_restarts), "TORCHELASTIC_RUN_ID": spec.rdzv_handler.get_run_id(), "TORCHELASTIC_USE_AGENT_STORE": str(use_agent_store), - "NCCL_ASYNC_ERROR_HANDLING": os.getenv( - "NCCL_ASYNC_ERROR_HANDLING", str(1) - ), + "NCCL_ASYNC_ERROR_HANDLING": os.getenv("NCCL_ASYNC_ERROR_HANDLING", + str(1)), } worker_env_ds.update(worker_env_elastic) if "OMP_NUM_THREADS" in os.environ: @@ -113,8 +112,6 @@ def _start_workers(self, worker_group: WorkerGroup) -> Dict[int, Any]: return self._pcontext.pids() - - def _invoke_run(self, role: str = "default") -> RunResult: # NOTE: currently only works for a single role @@ -122,8 +119,7 @@ def _invoke_run(self, role: str = "default") -> RunResult: role = spec.role log.info( - f"[{role}] starting workers for entrypoint: {spec.get_entrypoint_name()}" - ) + f"[{role}] starting workers for entrypoint: {spec.get_entrypoint_name()}") self._initialize_workers(self._worker_group) monitor_interval = spec.monitor_interval @@ -139,11 +135,12 @@ def _invoke_run(self, role: str = "default") -> RunResult: self._worker_group.state = state expire_time = datetime.utcnow() - ( - rdzv_handler._settings.keep_alive_interval * rdzv_handler._settings.keep_alive_max_attempt - ) + rdzv_handler._settings.keep_alive_interval * + rdzv_handler._settings.keep_alive_max_attempt) _dead_nodes = [ - node - for node, last_heartbeat in rdzv_handler._state_holder.state.last_heartbeats.items() + node for node, + last_heartbeat in + rdzv_handler._state_holder.state.last_heartbeats.items() if last_heartbeat < expire_time ] @@ -157,18 +154,20 @@ def _invoke_run(self, role: str = "default") -> RunResult: ) self._exit_barrier() return run_result - elif state in {WorkerState.UNHEALTHY, WorkerState.FAILED} or len(participants)>len( rdzv_handler._state_holder.state.participants): + elif state in { + WorkerState.UNHEALTHY, + WorkerState.FAILED + } or len(participants) > len(rdzv_handler._state_holder.state.participants): if self._remaining_restarts > 0: log.info( f"[{role}] Worker group {state.name}. " f"{self._remaining_restarts}/{spec.max_restarts} attempts left;" - f" will restart worker group" - ) + f" will restart worker group") self._remaining_restarts -= 1 # rdzv_handler._state_holder.state.restart = False self._restart_workers(self._worker_group) - participants = rdzv_handler._state_holder.state.participants - + participants = rdzv_handler._state_holder.state.participants + else: self._stop_workers(self._worker_group) self._worker_group.state = WorkerState.FAILED @@ -179,12 +178,10 @@ def _invoke_run(self, role: str = "default") -> RunResult: num_nodes_waiting = rdzv_handler.num_nodes_waiting() group_rank = self._worker_group.group_rank if num_nodes_waiting > 0: - log.info( - f"[{role}] Detected {num_nodes_waiting} " - f"new nodes from group_rank={group_rank}; " - f"will restart worker group" - ) + log.info(f"[{role}] Detected {num_nodes_waiting} " + f"new nodes from group_rank={group_rank}; " + f"will restart worker group") self._restart_workers(self._worker_group) - participants = rdzv_handler._state_holder.state.participants + participants = rdzv_handler._state_holder.state.participants else: raise Exception(f"[{role}] Worker group in {state.name} state") diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 23ad54bf3194..a8a776c4c317 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -25,8 +25,8 @@ from torch.distributed.elastic.rendezvous import RendezvousParameters from torch.distributed.elastic.agent.server.api import WorkerSpec import torch.distributed.elastic.rendezvous.registry as rdzv_registry -PID_FILE_BASEPATH = "/tmp" +PID_FILE_BASEPATH = "/tmp" def parse_args(): @@ -109,7 +109,7 @@ def parse_args(): def get_config_elastic(args, num_local_procs, node_rank) -> LaunchConfig: config: Dict[str, str] = {'timeout': 100} - config["store_type "] = "file" + config["store_type "] = "file" config = LaunchConfig( min_nodes=args.nnodes, @@ -129,12 +129,11 @@ def get_config_elastic(args, num_local_procs, node_rank) -> LaunchConfig: ) return config + def main(): args = parse_args() current_env = os.environ.copy() - - for k in current_env.keys(): if "NCCL" in k: logger.info(f"{args.node_rank} {k}={current_env[k]}") @@ -206,7 +205,7 @@ def main(): else: if args.module: raise ValueError("Don't use both the '--no_python' flag" - " and the '--module' flag at the same time.") + " and the '--module' flag at the same time.") cmd.append(args.training_script) # A user may not want to pass local_rank as a keyword arg so we make this optional. if not args.no_local_rank: @@ -216,12 +215,12 @@ def main(): process = subprocess.Popen(cmd, env=current_env) processes.append(process) else: - if args.min_nodes== -1: + if args.min_nodes == -1: args.min_nodes = 1 - if args.max_nodes== -1: + if args.max_nodes == -1: args.max_nodes = args.nnodes assert args.max_nodes > 0 and args.min_nodes > 0 , "Max and Min nodes should be positive" - + current_env["NCCL_ASYNC_ERROR_HANDLING"] = str(1) # Get config and arguments @@ -233,42 +232,37 @@ def main(): else: if args.module: raise ValueError("Don't use both the '--no_python' flag" - " and the '--module' flag at the same time.") + " and the '--module' flag at the same time.") cmd.append(args.training_script) cmd += args.training_script_args - elastic_config = get_config_elastic(args, num_local_procs,args.node_rank) + elastic_config = get_config_elastic(args, num_local_procs, args.node_rank) cmd_args = cmd[1:] rdzv_configs: Dict[str, str] = {'timeout': 100} - rdzv_parameters = RendezvousParameters( - backend='c10d', - endpoint=args.master_addr+":"+str(args.master_port), - run_id='123456789', - min_nodes=args.min_nodes, - max_nodes=args.max_nodes, - **rdzv_configs - ) + rdzv_parameters = RendezvousParameters(backend='c10d', + endpoint=args.master_addr + ":" + + str(args.master_port), + run_id='123456789', + min_nodes=args.min_nodes, + max_nodes=args.max_nodes, + **rdzv_configs) spec = WorkerSpec( - role='trainer', - local_world_size=num_local_procs, - entrypoint=cmd[0], - args=cmd[1:], - rdzv_handler=rdzv_registry.get_rendezvous_handler(rdzv_parameters), - max_restarts=100, - monitor_interval=5, - redirects=Std.from_str("0"), - tee=Std.from_str("0"), - master_addr=None, - master_port=None, - ) - agent = DSElasticAgent( - spec, - current_env + role='trainer', + local_world_size=num_local_procs, + entrypoint=cmd[0], + args=cmd[1:], + rdzv_handler=rdzv_registry.get_rendezvous_handler(rdzv_parameters), + max_restarts=100, + monitor_interval=5, + redirects=Std.from_str("0"), + tee=Std.from_str("0"), + master_addr=None, + master_port=None, ) + agent = DSElasticAgent(spec, current_env) agent.run() - sig_names = {2: "SIGINT", 15: "SIGTERM"} last_return_code = None diff --git a/deepspeed/launcher/runner.py b/deepspeed/launcher/runner.py index 58b38d22017a..ffe85a142d7d 100755 --- a/deepspeed/launcher/runner.py +++ b/deepspeed/launcher/runner.py @@ -329,6 +329,7 @@ def run_autotuning(args, active_resources): if args.autotuning == "run": tuner.run_after_tuning() + def parse_num_nodes(str_num_nodes: str, elastic_training: bool): node_list = str_num_nodes.split(":") @@ -340,7 +341,7 @@ def parse_num_nodes(str_num_nodes: str, elastic_training: bool): raise RuntimeError("MIN:MAX format is only supported in elastic training") else: raise RuntimeError("num_nodes {} is not in MIN:MAX format".format(str_num_nodes)) - + return min_nodes, max_nodes @@ -348,7 +349,7 @@ def main(args=None): args = parse_args(args) if args.elastic_training: - assert args.master_addr != "", "Master Addr is required when elastic training is enabled" + assert args.master_addr != "", "Master Addr is required when elastic training is enabled" resource_pool = fetch_hostfile(args.hostfile) From f6375690b6f67aaac6751634dfb689f65a70bbe5 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Wed, 6 Jul 2022 03:11:25 +0500 Subject: [PATCH 12/27] add launch.py and elastic_agent.py files to skip list in torchdist check --- .pre-commit-config.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b68175b8272a..c5c7e39822d1 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -39,7 +39,7 @@ repos: name: check-torchdist entry: ./scripts/check-torchdist.py language: script - exclude: ^(deepspeed/comm/|docs/|benchmarks/|scripts/check-torchdist.py|deepspeed/moe/sharded_moe.py|deepspeed/runtime/comm/coalesced_collectives.py) + exclude: ^(deepspeed/comm/|docs/|benchmarks/|scripts/check-torchdist.py|deepspeed/moe/sharded_moe.py|deepspeed/runtime/comm/coalesced_collectives.py|deepspeed/elasticity/elastic_agent.py|deepspeed/launcher/launch.py) # Specific deepspeed/ files are excluded for now until we wrap ProcessGroup in deepspeed.comm - repo: https://github.com/codespell-project/codespell From 8aafc3a5e97afe73be9d849180f94506776a3a9b Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 7 Jul 2022 02:16:27 +0500 Subject: [PATCH 13/27] add pytorch dependency for elastic training --- deepspeed/elasticity/__init__.py | 5 +++- deepspeed/elasticity/utils.py | 0 deepspeed/launcher/launch.py | 45 ++++++++++---------------------- 3 files changed, 18 insertions(+), 32 deletions(-) create mode 100644 deepspeed/elasticity/utils.py diff --git a/deepspeed/elasticity/__init__.py b/deepspeed/elasticity/__init__.py index 6635614a8dca..32627dbbe7c1 100644 --- a/deepspeed/elasticity/__init__.py +++ b/deepspeed/elasticity/__init__.py @@ -1,2 +1,5 @@ from .elasticity import compute_elastic_config, elasticity_enabled, ensure_immutable_elastic_config -from .elastic_agent import DSElasticAgent +from .utils import is_torch_elastic_compatible + +if is_torch_elastic_compatible(): + from .elastic_agent import DSElasticAgent diff --git a/deepspeed/elasticity/utils.py b/deepspeed/elasticity/utils.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index d35cf71db716..bc915690acd3 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -17,14 +17,9 @@ import signal from collections import defaultdict from argparse import ArgumentParser, REMAINDER -from torch.distributed.launcher.api import LaunchConfig, elastic_launch -from torch.distributed.elastic.multiprocessing import Std from ..constants import TORCH_DISTRIBUTED_DEFAULT_PORT from ..utils import logger -from ..elasticity import DSElasticAgent -from torch.distributed.elastic.rendezvous import RendezvousParameters -from torch.distributed.elastic.agent.server.api import WorkerSpec -import torch.distributed.elastic.rendezvous.registry as rdzv_registry +from ..elasticity import is_torch_elastic_compatible PID_FILE_BASEPATH = "/tmp" @@ -106,30 +101,6 @@ def parse_args(): return parser.parse_args() -def get_config_elastic(args, num_local_procs, node_rank) -> LaunchConfig: - - config: Dict[str, str] = {'timeout': 100} - config["store_type "] = "file" - - config = LaunchConfig( - min_nodes=args.nnodes, - max_nodes=args.max_nodes, - nproc_per_node=num_local_procs, - run_id="123456789", - role="default", - rdzv_endpoint="worker-0:46728", - rdzv_backend='c10d', - rdzv_configs=config, - max_restarts=100, - monitor_interval=1, - start_method="spawn", - redirects=Std.from_str("0"), - tee=Std.from_str("0"), - log_dir="", - ) - return config - - def main(): args = parse_args() current_env = os.environ.copy() @@ -186,6 +157,13 @@ def main(): with open(pid_file, 'w') as fd: fd.write(f"{launcher_pid}") + if not is_torch_elastic_compatible(): + if args.enable_elastic_training: + logger.info(f"Disabling elastic training support as \ + PyTorch version should be greater than 1.11.x.\ + Current version is {torch.__version__}") + args.enable_elastic_training = False + processes = [] cmd = [] @@ -215,6 +193,12 @@ def main(): process = subprocess.Popen(cmd, env=current_env) processes.append(process) else: + from ..elasticity import DSElasticAgent + from torch.distributed.elastic.rendezvous import RendezvousParameters + from torch.distributed.elastic.agent.server.api import WorkerSpec + import torch.distributed.elastic.rendezvous.registry as rdzv_registry + from torch.distributed.elastic.multiprocessing import Std + if args.min_nodes == -1: args.min_nodes = 1 if args.max_nodes == -1: @@ -235,7 +219,6 @@ def main(): " and the '--module' flag at the same time.") cmd.append(args.training_script) cmd += args.training_script_args - elastic_config = get_config_elastic(args, num_local_procs, args.node_rank) cmd_args = cmd[1:] rdzv_configs: Dict[str, str] = {'timeout': 100} From b0a8802dce2bab8737cfc3ead021995177eef561 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 7 Jul 2022 02:37:11 +0500 Subject: [PATCH 14/27] add function for checking pytorch version --- deepspeed/elasticity/utils.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/deepspeed/elasticity/utils.py b/deepspeed/elasticity/utils.py index e69de29bb2d1..a1001c6b3806 100644 --- a/deepspeed/elasticity/utils.py +++ b/deepspeed/elasticity/utils.py @@ -0,0 +1,14 @@ +import torch + + +def is_torch_elastic_compatible(): + ''' + Helper to lookup torch version. Elastic training is + introduced in 1.11.x + ''' + TORCH_MAJOR = int(torch.__version__.split('.')[0]) + TORCH_MINOR = int(torch.__version__.split('.')[1]) + if TORCH_MAJOR == 1 and TORCH_MINOR >= 11: + return True + else: + return False From 96d678bd81a4977bb795ecf70cdae5ddacb0408d Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Sun, 10 Jul 2022 11:08:15 +0500 Subject: [PATCH 15/27] added kill command for pdsh when SIGINT is received --- deepspeed/launcher/multinode_runner.py | 6 +++++- deepspeed/launcher/runner.py | 20 ++++++++++++++++++-- 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/deepspeed/launcher/multinode_runner.py b/deepspeed/launcher/multinode_runner.py index d1cbe5e9bcd7..1b9159034bd0 100644 --- a/deepspeed/launcher/multinode_runner.py +++ b/deepspeed/launcher/multinode_runner.py @@ -98,8 +98,12 @@ def get_cmd(self, environment, active_resources): deepspeed_launch.append("--enable_elastic_training") deepspeed_launch.append(f"--max_nodes={self.args.max_num_nodes}") deepspeed_launch.append(f"--min_nodes={self.args.min_num_nodes}") + + cmd_to_search = [i + "\\" for i in deepspeed_launch[2:6]] + + kill_command = pdsh_cmd_args + ["pkill -f ", " ".join(cmd_to_search)[:-2]] return pdsh_cmd_args + deepspeed_launch + [self.user_script - ] + self.user_arguments + ] + self.user_arguments, kill_command class OpenMPIRunner(MultiNodeRunner): diff --git a/deepspeed/launcher/runner.py b/deepspeed/launcher/runner.py index ffe85a142d7d..84514aa6aa0b 100755 --- a/deepspeed/launcher/runner.py +++ b/deepspeed/launcher/runner.py @@ -14,7 +14,8 @@ import subprocess import collections from copy import deepcopy - +import signal +import time import torch.cuda from .multinode_runner import PDSHRunner, OpenMPIRunner, MVAPICHRunner @@ -490,11 +491,26 @@ def main(args=None): key, val = var.split('=', maxsplit=1) runner.add_export(key, val) - cmd = runner.get_cmd(env, active_resources) + if args.launcher == PDSH_LAUNCHER: + cmd, kill_cmd = runner.get_cmd(env, active_resources) + else: + cmd = runner.get_cmd(env, active_resources) logger.info(f"cmd = {' '.join(cmd)}") result = subprocess.Popen(cmd, env=env) + def sigkill_handler(signum, frame): + result.send_signal(signal.SIGINT) + time.sleep(0.1) + result.send_signal(signal.SIGTERM) + result_kill = subprocess.Popen(kill_cmd, env=env) + result_kill.wait() + time.sleep(1) + sys.exit(1) + + if args.launcher == PDSH_LAUNCHER: + signal.signal(signal.SIGINT, sigkill_handler) + result.wait() # In case of failure must propagate the error-condition back to the caller (usually shell). The From 3ac51197ce1b168ee44119f4ca0e7d1c605bcb94 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 15 Jul 2022 03:34:59 +0500 Subject: [PATCH 16/27] re-enable elastic checkpoint assertion --- deepspeed/runtime/engine.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/deepspeed/runtime/engine.py b/deepspeed/runtime/engine.py index 1ce904a5c770..892394e049ad 100644 --- a/deepspeed/runtime/engine.py +++ b/deepspeed/runtime/engine.py @@ -2649,8 +2649,7 @@ def _load_zero_checkpoint(self, load_dir, tag, load_optimizer_states=True): if zero_sd_list is None: return False - if load_optimizer_states and self.dp_world_size != self.loaded_checkpoint_dp_world_size \ - and not self.zero_elastic_checkpoint(): + if load_optimizer_states and self.dp_world_size != self.loaded_checkpoint_dp_world_size: raise ZeRORuntimeException("The checkpoint being loaded used a DP " \ f"world size of {self.loaded_checkpoint_dp_world_size} but the " \ f"current world size is {self.dp_world_size}. Automatic adjustment " \ From 4f9c53513a6abd8275600021df0501a92f5eb9fe Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Tue, 19 Jul 2022 15:33:04 +0000 Subject: [PATCH 17/27] Add support for variable batch size --- deepspeed/elasticity/__init__.py | 2 +- deepspeed/elasticity/config.py | 15 ++++ deepspeed/elasticity/constants.py | 8 ++- deepspeed/elasticity/elasticity.py | 106 +++++++++++++++++++++++++++-- deepspeed/runtime/config.py | 19 ++++++ deepspeed/runtime/engine.py | 16 ++++- 6 files changed, 157 insertions(+), 9 deletions(-) diff --git a/deepspeed/elasticity/__init__.py b/deepspeed/elasticity/__init__.py index 32627dbbe7c1..1c3b710fb171 100644 --- a/deepspeed/elasticity/__init__.py +++ b/deepspeed/elasticity/__init__.py @@ -1,5 +1,5 @@ from .elasticity import compute_elastic_config, elasticity_enabled, ensure_immutable_elastic_config from .utils import is_torch_elastic_compatible - +from .constants import ENABLED, ENABLED_DEFAULT if is_torch_elastic_compatible(): from .elastic_agent import DSElasticAgent diff --git a/deepspeed/elasticity/config.py b/deepspeed/elasticity/config.py index 67db58c70e71..ea03a7c8a4ad 100644 --- a/deepspeed/elasticity/config.py +++ b/deepspeed/elasticity/config.py @@ -79,6 +79,7 @@ def __init__(self, param_dict): self.min_gpus = param_dict.get(MIN_GPUS, MIN_GPUS_DEFAULT) self.max_gpus = param_dict.get(MAX_GPUS, MAX_GPUS_DEFAULT) + if self.min_gpus < 1 or self.max_gpus < 1: raise ElasticityConfigError( "Elasticity min/max gpus must be > 0, " @@ -88,6 +89,20 @@ def __init__(self, param_dict): "Elasticity min_gpus cannot be greater than max_gpus, " f"given min_gpus: {self.min_gpus}, max_gpus: {self.max_gpus}") + self.model_parallel_size = param_dict.get(MODEL_PARLLEL_SIZE, + MODEL_PARLLEL_SIZE_DEFAULT) + if self.model_parallel_size < 1: + raise ElasticityConfigError( + "Model-Parallel size cannot be less than 1, " + f"given model-parallel size: {self.model_parallel_size}") + + self.num_gpus_per_node = param_dict.get(NUM_GPUS_PER_NODE, + NUM_GPUS_PER_NODE_DEFAULT) + if self.num_gpus_per_node < 1: + raise ElasticityConfigError( + "NUmber of GPUs per node cannot be less than 1, " + f"given number of GPUs per node: {self.num_gpus_per_node}") + self.min_time = param_dict.get(MIN_TIME, MIN_TIME_DEFAULT) if self.min_time < 0: raise ElasticityConfigError( diff --git a/deepspeed/elasticity/constants.py b/deepspeed/elasticity/constants.py index cf428b70a0f4..eb40edc84711 100644 --- a/deepspeed/elasticity/constants.py +++ b/deepspeed/elasticity/constants.py @@ -27,7 +27,7 @@ ELASTICITY = 'elasticity' # Current elasticity version -LATEST_ELASTICITY_VERSION = 0.1 +LATEST_ELASTICITY_VERSION = 0.2 ENABLED = 'enabled' ENABLED_DEFAULT = False @@ -46,6 +46,12 @@ MAX_GPUS = 'max_gpus' MAX_GPUS_DEFAULT = 10000 +NUM_GPUS_PER_NODE = 'num_gpus_per_node' +NUM_GPUS_PER_NODE_DEFAULT = 1 + +MODEL_PARLLEL_SIZE = "model_parallel_size" +MODEL_PARLLEL_SIZE_DEFAULT = 1 + # Minimum running time (minutes) before the scheduler will scale us, 0 implies it's unknown MIN_TIME = "min_time" MIN_TIME_DEFAULT = 0 diff --git a/deepspeed/elasticity/elasticity.py b/deepspeed/elasticity/elasticity.py index e678d5ed836e..8459414a5b3a 100644 --- a/deepspeed/elasticity/elasticity.py +++ b/deepspeed/elasticity/elasticity.py @@ -5,7 +5,7 @@ import re import json import numpy as np - +import math from packaging import version as pkg_version from .config import ElasticityConfig, ElasticityConfigError, ElasticityError, \ @@ -15,7 +15,7 @@ IGNORE_NON_ELASTIC_BATCH_INFO_DEFAULT, DEEPSPEED_ELASTICITY_CONFIG from ..git_version_info import version as __version__ from ..utils import logger - +import subprocess # Thirty eight smallest highly composite numbers. The list should # be enough to support up to 720K batch size. HCN_LIST = [ @@ -93,7 +93,6 @@ def get_valid_gpus(batch_size, micro_batches, min_valid_gpus, max_valid_gpus): valid_gpus.append(i) valid_gpus = set(valid_gpus) valid_gpus = sorted(list(valid_gpus)) - logger.info(f"Valid GPUs: {valid_gpus}") return valid_gpus @@ -173,6 +172,68 @@ def _get_compatible_gpus_v01(micro_batches, return final_batch_size, valid_gpus +def _get_compatible_gpus_v02(micro_batches, + max_acceptable_batch_size, + current_num_gpus, + min_gpus=None, + max_gpus=None, + prefer_larger=True, + num_gpus_per_node=1, + model_parallel_size=1): + ''' + Returns: + final_batch_size + valid_gpus + micro-batch size + ''' + assert num_gpus_per_node % model_parallel_size == 0, \ + f"In Elasticity v0.2, number of GPUs per node:" \ + f"{num_gpus_per_node} should be divisible by" \ + f"model parallel size {model_parallel_size}" + + def get_microbatch(final_batch_size): + candidate_microbatch = None + + for micro_batch in micro_batches: + if final_batch_size // current_num_gpus % micro_batch == 0: + if candidate_microbatch == None: + candidate_microbatch = micro_batch + if prefer_larger and candidate_microbatch < micro_batch: + candidate_microbatch = micro_batch + return candidate_microbatch + + dp_size_per_node = num_gpus_per_node // model_parallel_size + + final_batch_size, valid_world_size = _get_compatible_gpus_v01(micro_batches, + int(max_acceptable_batch_size/dp_size_per_node), + int(min_gpus/num_gpus_per_node), + int(max_gpus/num_gpus_per_node), # Passing number of max nodes as Elasticity v2 works at node level + prefer_larger=prefer_larger) + + final_batch_size = int(final_batch_size) * dp_size_per_node + + if current_num_gpus // dp_size_per_node in valid_world_size: + candidate_microbatch = get_microbatch(final_batch_size) + return final_batch_size, valid_world_size, candidate_microbatch + + current_dp_size = (current_num_gpus / num_gpus_per_node) * dp_size_per_node + candidate_batch_sizes = [] + for micro_batch in micro_batches: + min_batch_size = micro_batch * current_dp_size + + factor = math.floor(max_acceptable_batch_size / float(min_batch_size)) + candidate_batch_sizes.append(factor * min_batch_size) + + used_microbatch = None + if prefer_larger: + candidate_batch_size = max(candidate_batch_sizes) + else: + candidate_batch_size = min(candidate_batch_sizes) + candidate_microbatch = get_microbatch(candidate_batch_size) + + return candidate_batch_size, [current_dp_size], candidate_microbatch + + def _compatible_ds_version_check(target_deepspeed_version: str): min_version = pkg_version.parse(MINIMUM_DEEPSPEED_VERSION) target_version = pkg_version.parse(target_deepspeed_version) @@ -223,7 +284,10 @@ def ensure_immutable_elastic_config(runtime_elastic_config_dict: dict): "guarantee resource scheduler will scale this job using compatible GPU counts.") -def compute_elastic_config(ds_config: dict, target_deepspeed_version: str, world_size=0): +def compute_elastic_config(ds_config: dict, + target_deepspeed_version: str, + world_size=0, + return_microbatch=False): """Core deepspeed elasticity API. Given an elastic config (similar to the example below) DeepSpeed will compute a total train batch size corresponding valid GPU count list that provides a high level of elasticity. Elasticity in this case means we are safe to scale @@ -252,6 +316,7 @@ def compute_elastic_config(ds_config: dict, target_deepspeed_version: str, world compatible with the elasticity version used in the backend. world_size (int, optional): Intended/current world size, will do some sanity checks to ensure world size is actually valid with the config. + return_microbatch (bool, optional): whether to return micro batch size or not. Raises: ElasticityConfigError: Missing required elasticity config or elasticity disabled @@ -277,6 +342,8 @@ def compute_elastic_config(ds_config: dict, target_deepspeed_version: str, world "('enabled':true) if running an elastic training job.") elastic_config = ElasticityConfig(elastic_config_dict) + model_parallel_size = elastic_config.model_parallel_size + num_gpus_per_node = elastic_config.num_gpus_per_node if float(elastic_config.version) > LATEST_ELASTICITY_VERSION: raise ElasticityConfigError("Attempting to run elasticity version " \ @@ -297,10 +364,26 @@ def compute_elastic_config(ds_config: dict, target_deepspeed_version: str, world prefer_larger=elastic_config.prefer_larger_batch_size) # ensure batch size is int dtype final_batch_size = int(final_batch_size) + elif float(elastic_config.version) == 0.2: + current_num_gpus = int(os.getenv('WORLD_SIZE')) + + final_batch_size, valid_gpus, candidate_microbatch_size = _get_compatible_gpus_v02( + micro_batches=elastic_config.micro_batches, + max_acceptable_batch_size=elastic_config.max_acceptable_batch_size, + current_num_gpus=current_num_gpus, + min_gpus=elastic_config.min_gpus, + max_gpus=elastic_config.max_gpus, + prefer_larger=elastic_config.prefer_larger_batch_size, + num_gpus_per_node=num_gpus_per_node, + model_parallel_size=model_parallel_size) + # ensure batch size is int dtype + final_batch_size = int(final_batch_size) else: raise NotImplementedError( f"Unable to find elastic logic for version: {elastic_config.version}") + logger.info(f"Valid World Size (GPUs / Model Parallel Size): {valid_gpus}") + if world_size > 0: if world_size not in valid_gpus: raise ElasticityIncompatibleWorldSize(f"World size ({world_size}) is not valid " \ @@ -317,4 +400,19 @@ def compute_elastic_config(ds_config: dict, target_deepspeed_version: str, world f" micro_batches={elastic_config.micro_batches}." return final_batch_size, valid_gpus, micro_batch_size + if return_microbatch: + # Pick a valid micro batch size + if float(elastic_config.version) == 0.2: + return final_batch_size, valid_gpus, candidate_microbatch_size + else: + micro_batch_size = None + for mbsz in sorted(list(set(elastic_config.micro_batches)), reverse=True): + if final_batch_size // world_size % mbsz == 0: + micro_batch_size = mbsz + break + assert micro_batch_size is not None, "Unable to find divisible micro batch size" \ + f" world_size={world_size}, final_batch_size={final_batch_size}, and " \ + f" micro_batches={elastic_config.micro_batches}." + return final_batch_size, valid_gpus, micro_batch_size + return final_batch_size, valid_gpus diff --git a/deepspeed/runtime/config.py b/deepspeed/runtime/config.py index 4571cbdf7056..300163cbb0e3 100755 --- a/deepspeed/runtime/config.py +++ b/deepspeed/runtime/config.py @@ -41,6 +41,10 @@ ELASTICITY, IGNORE_NON_ELASTIC_BATCH_INFO, IGNORE_NON_ELASTIC_BATCH_INFO_DEFAULT, + MODEL_PARLLEL_SIZE, + MODEL_PARLLEL_SIZE_DEFAULT, + NUM_GPUS_PER_NODE, + NUM_GPUS_PER_NODE_DEFAULT, ) from ..profiling.config import DeepSpeedFlopsProfilerConfig @@ -801,6 +805,21 @@ def __init__(self, config: Union[str, dict], mpu=None): # Ensure the resource scheduler saw the same elastic config we are using at runtime ensure_immutable_elastic_config(runtime_elastic_config_dict=elastic_dict) + self.elastic_model_parallel_size = elastic_dict.get( + MODEL_PARLLEL_SIZE, + MODEL_PARLLEL_SIZE_DEFAULT) + if self.elastic_model_parallel_size < 1: + raise ElasticityConfigError( + "Model-Parallel size cannot be less than 1, " + f"given model-parallel size: {self.elastic_model_parallel_size}") + + self.num_gpus_per_node = elastic_dict.get(NUM_GPUS_PER_NODE, + NUM_GPUS_PER_NODE_DEFAULT) + if self.num_gpus_per_node < 1: + raise ElasticityConfigError( + "NUmber of GPUs per node cannot be less than 1, " + f"given number of GPUs per node: {self.num_gpus_per_node}") + ignore_non_elastic_batch_info = elastic_dict.get( IGNORE_NON_ELASTIC_BATCH_INFO, IGNORE_NON_ELASTIC_BATCH_INFO_DEFAULT) diff --git a/deepspeed/runtime/engine.py b/deepspeed/runtime/engine.py index 6fd84a6aa073..b3225851430d 100644 --- a/deepspeed/runtime/engine.py +++ b/deepspeed/runtime/engine.py @@ -258,9 +258,11 @@ def __init__( see_memory_usage(f"DeepSpeed Engine: After args sanity test", force=self.memory_breakdown()) if mpu is not None: - assert not self.elasticity_enabled(), ( - "Elasticity is not currently supported" " with model parallelism." - ) + if self.elasticity_enabled(): + if not self.is_elastic_model_parallel_supported(): + assert not self.elasticity_enabled(), ( + "Elasticity is not currently supported" " with model parallelism." + ) self._set_distributed_vars(args) @@ -463,6 +465,14 @@ def checkpoint_tag_validation_fail(self): def elasticity_enabled(self): return self._config.elasticity_enabled + def is_elastic_model_parallel_supported(self): + if self.elasticity_enabled(): + # Add code for finding number of GPUs per node automatically + if self._config.num_gpus_per_node % self._config.elastic_model_parallel_size == 0: + return True + else: + return False + def pld_enabled(self): return self._config.pld_enabled From d995fb3beceab46426a9f225996079c417f21577 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Sat, 23 Jul 2022 00:22:54 +0500 Subject: [PATCH 18/27] Fix elasticity V2, enable pipeline parallelism in Elastic Training, and add ENABLE constants to init --- deepspeed/elasticity/__init__.py | 2 +- deepspeed/elasticity/elasticity.py | 9 +++++---- deepspeed/runtime/pipe/engine.py | 6 ++++-- 3 files changed, 10 insertions(+), 7 deletions(-) diff --git a/deepspeed/elasticity/__init__.py b/deepspeed/elasticity/__init__.py index 1c3b710fb171..6ad9f20cf936 100644 --- a/deepspeed/elasticity/__init__.py +++ b/deepspeed/elasticity/__init__.py @@ -1,5 +1,5 @@ from .elasticity import compute_elastic_config, elasticity_enabled, ensure_immutable_elastic_config from .utils import is_torch_elastic_compatible -from .constants import ENABLED, ENABLED_DEFAULT +from .constants import ENABLED, ENABLED_DEFAULT, ELASTICITY if is_torch_elastic_compatible(): from .elastic_agent import DSElasticAgent diff --git a/deepspeed/elasticity/elasticity.py b/deepspeed/elasticity/elasticity.py index 8459414a5b3a..f10daec4db2f 100644 --- a/deepspeed/elasticity/elasticity.py +++ b/deepspeed/elasticity/elasticity.py @@ -211,10 +211,10 @@ def get_microbatch(final_batch_size): prefer_larger=prefer_larger) final_batch_size = int(final_batch_size) * dp_size_per_node - - if current_num_gpus // dp_size_per_node in valid_world_size: + valid_dp_world_size = [i * dp_size_per_node for i in valid_world_size] + if current_num_gpus // model_parallel_size in valid_dp_world_size: candidate_microbatch = get_microbatch(final_batch_size) - return final_batch_size, valid_world_size, candidate_microbatch + return final_batch_size, valid_dp_world_size, candidate_microbatch current_dp_size = (current_num_gpus / num_gpus_per_node) * dp_size_per_node candidate_batch_sizes = [] @@ -229,9 +229,10 @@ def get_microbatch(final_batch_size): candidate_batch_size = max(candidate_batch_sizes) else: candidate_batch_size = min(candidate_batch_sizes) + candidate_microbatch = get_microbatch(candidate_batch_size) - return candidate_batch_size, [current_dp_size], candidate_microbatch + return candidate_batch_size, [int(current_dp_size)], candidate_microbatch def _compatible_ds_version_check(target_deepspeed_version: str): diff --git a/deepspeed/runtime/pipe/engine.py b/deepspeed/runtime/pipe/engine.py index 94add6f9c8e4..33c2edd05e35 100644 --- a/deepspeed/runtime/pipe/engine.py +++ b/deepspeed/runtime/pipe/engine.py @@ -80,8 +80,10 @@ def __init__(self, has_bool_tensors=False, *super_args, **super_kwargs): # used to disable the pipeline all-reduce when used with 1-bit Adam/1-bit LAMB self.pipeline_enable_backward_allreduce = True - assert not self.elasticity_enabled(), "Elasticity is not currently supported" \ - " with pipeline parallelism." + if self.elasticity_enabled(): + if not self.is_elastic_model_parallel_supported(): + assert not self.elasticity_enabled(), "Elasticity is not currently supported" \ + " with pipeline parallelism." # pipeline step for logging self.log_batch_step_id = -1 From 706ebcece01df351eacaa26ede879647d344eee3 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Sat, 23 Jul 2022 00:27:00 +0500 Subject: [PATCH 19/27] updated elastic unit test --- tests/unit/test_elastic.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_elastic.py b/tests/unit/test_elastic.py index 353d6def37ba..753acec08921 100644 --- a/tests/unit/test_elastic.py +++ b/tests/unit/test_elastic.py @@ -78,7 +78,7 @@ def test_invalid_world_size(): def test_future_elastic_version(): ds_config = base_ds_config.copy() - ds_config['elasticity']['version'] = '0.2' + ds_config['elasticity']['version'] = '0.3' with pytest.raises(deepspeed.elasticity.config.ElasticityError): deepspeed.elasticity.compute_elastic_config(ds_config=ds_config, target_deepspeed_version=ds_version) From 9063a948bd0a2c9da2392156b9bf7e51c1ca3896 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Sat, 23 Jul 2022 12:22:22 +0500 Subject: [PATCH 20/27] added an assertion for moded-parallel support and added code to protect WORLD_SIZE computation --- deepspeed/elasticity/elasticity.py | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/deepspeed/elasticity/elasticity.py b/deepspeed/elasticity/elasticity.py index f10daec4db2f..a6679c93e86b 100644 --- a/deepspeed/elasticity/elasticity.py +++ b/deepspeed/elasticity/elasticity.py @@ -315,7 +315,7 @@ def compute_elastic_config(ds_config: dict, target_deepspeed_version (str): When called from scheduling infrastructure we want to ensure that the target deepspeed version is compatible with the elasticity version used in the backend. - world_size (int, optional): Intended/current world size, will do some sanity + world_size (int, optional): Intended/current DP world size, will do some sanity checks to ensure world size is actually valid with the config. return_microbatch (bool, optional): whether to return micro batch size or not. @@ -346,6 +346,11 @@ def compute_elastic_config(ds_config: dict, model_parallel_size = elastic_config.model_parallel_size num_gpus_per_node = elastic_config.num_gpus_per_node + if model_parallel_size > 1 and float(elastic_config.version) != 0.2: + raise ElasticityConfigError(f"Elasticity V{elastic_config.version} " \ + f"does not support model-parallel training. Given model-parallel size: " \ + f"{model_parallel_size}") + if float(elastic_config.version) > LATEST_ELASTICITY_VERSION: raise ElasticityConfigError("Attempting to run elasticity version " \ f"{elastic_config.version} but runtime only supports up " \ @@ -366,7 +371,20 @@ def compute_elastic_config(ds_config: dict, # ensure batch size is int dtype final_batch_size = int(final_batch_size) elif float(elastic_config.version) == 0.2: - current_num_gpus = int(os.getenv('WORLD_SIZE')) + if world_size != 0 and False: + current_num_gpus = world_size + else: + if "WORLD_SIZE" in os.environ and \ + os.getenv('WORLD_SIZE').isnumeric(): + current_num_gpus = int(os.getenv('WORLD_SIZE')) + else: + WORLD_SIZE = os.getenv('WORLD_SIZE') + raise ElasticityConfigError( + 'Elasticity V 0.2 needs WORLD_SIZE '\ + 'to compute valid batch size. '\ + 'Either give it as argument to function compute_elastic_config '\ + 'or set it as an environment variable. '\ + f'Value of WORLD_SIZE as environment variable is {WORLD_SIZE}') final_batch_size, valid_gpus, candidate_microbatch_size = _get_compatible_gpus_v02( micro_batches=elastic_config.micro_batches, From f4ace715a57e7da51640260a0687c781297462fb Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 28 Jul 2022 21:40:32 +0000 Subject: [PATCH 21/27] modified elastic training unit test, added config options for elastic training in docs, and added an assertion in runner for elastic training --- deepspeed/elasticity/elasticity.py | 9 ++++---- deepspeed/launcher/constants.py | 2 ++ deepspeed/launcher/runner.py | 3 +++ docs/_pages/config-json.md | 31 +++++++++++++++++++++++++ tests/unit/test_elastic.py | 37 ++++++++++++++++++++++++++++++ 5 files changed, 78 insertions(+), 4 deletions(-) diff --git a/deepspeed/elasticity/elasticity.py b/deepspeed/elasticity/elasticity.py index a6679c93e86b..fcacb40c53cd 100644 --- a/deepspeed/elasticity/elasticity.py +++ b/deepspeed/elasticity/elasticity.py @@ -186,10 +186,11 @@ def _get_compatible_gpus_v02(micro_batches, valid_gpus micro-batch size ''' - assert num_gpus_per_node % model_parallel_size == 0, \ + if num_gpus_per_node % model_parallel_size != 0: + raise ElasticityError( f"In Elasticity v0.2, number of GPUs per node:" \ - f"{num_gpus_per_node} should be divisible by" \ - f"model parallel size {model_parallel_size}" + f"{num_gpus_per_node} should be divisible by " \ + f"model parallel size {model_parallel_size}") def get_microbatch(final_batch_size): candidate_microbatch = None @@ -371,7 +372,7 @@ def compute_elastic_config(ds_config: dict, # ensure batch size is int dtype final_batch_size = int(final_batch_size) elif float(elastic_config.version) == 0.2: - if world_size != 0 and False: + if world_size != 0: current_num_gpus = world_size else: if "WORLD_SIZE" in os.environ and \ diff --git a/deepspeed/launcher/constants.py b/deepspeed/launcher/constants.py index 15b262342d47..2ae22ea6c7da 100644 --- a/deepspeed/launcher/constants.py +++ b/deepspeed/launcher/constants.py @@ -7,3 +7,5 @@ MVAPICH_LAUNCHER = 'mvapich' MVAPICH_TMP_HOSTFILE = '/tmp/deepspeed_mvapich_hostfile' + +ELASTIC_TRAINING_ID_DEFAULT = "123456789" diff --git a/deepspeed/launcher/runner.py b/deepspeed/launcher/runner.py index 84514aa6aa0b..8da8469ffca5 100755 --- a/deepspeed/launcher/runner.py +++ b/deepspeed/launcher/runner.py @@ -430,6 +430,9 @@ def main(args=None): updated_active_resources[hostname] = list(range(args.num_gpus)) active_resources = updated_active_resources + if args.elastic_training: + assert not args.no_local_rank, "--no_local_rank argument is not supported in Elastic training" + # encode world info as base64 to make it easier to pass via command line world_info_base64 = encode_world_info(active_resources) diff --git a/docs/_pages/config-json.md b/docs/_pages/config-json.md index 3b283b459b18..4dec6a260ef2 100755 --- a/docs/_pages/config-json.md +++ b/docs/_pages/config-json.md @@ -1044,3 +1044,34 @@ Example of **csv_monitor** configuration: "job_name": "train_bert" } ``` + +### Elastic Training Config (V0.1 and V0.2) + +```json + "elasticity": { + "enabled": true, + "max_train_batch_size": "seqlen", + "micro_batch_sizes": 8, + "min_gpus": 1024, + "max_gpus": "fixed_linear", + "min_time": "seqlen", + "version": 8, + "ignore_non_elastic_batch_info": 1024, + "num_gpus_per_node": "fixed_linear", + "model_parallel_size": MODEL_PARALLEL_SIZE + } +``` + +| Field | Description |Default| +| ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ----- | +| `enabled` | Enables computation of global batch size in elastic training. | false | +| `max_train_batch_size` | Max acceptable batch size can be used in training. | 2000 | +| `micro_batch_sizes` | Acceptable micro batch sizes, same as train_micro_batch_size_per_gpu | [2,4,6] | +| `min_gpus` | Min number of GPUs to search over when computing highly composite batch size in v0.1 and v0.2. | 1 | +| `max_gpus` | Max number of GPUs to search over when computing highly composite batch size in v0.1 and v0.2. | 10000 | +| `min_time` |Minimum running time (minutes) before the scheduler will scale again (only used in v0.1). 0 implies it's unknown | 0 | +| `prefer_large_batch` | When finding a suitable batch size, attempt to find one that is closest to the max train batch size given. | true | +| `version` | Version of elastic logic to use. | 0.2 | +| `ignore_non_elastic_batch_info` | Ignore all batch info provided outside the elastic config. To reduce confusion, we require all batch related info to be given in elastic config only. | false | +| `num_gpus_per_node` | Number of GPUs per node. This information is used by v0.2 to support model-parallel training (only used by v0.2) | 1 | +| `model_parallel_size` | Tensor or model parallel size (only used by v0.2) | 1 | diff --git a/tests/unit/test_elastic.py b/tests/unit/test_elastic.py index 753acec08921..5c7e03da710d 100644 --- a/tests/unit/test_elastic.py +++ b/tests/unit/test_elastic.py @@ -3,6 +3,7 @@ from .common import distributed_test from deepspeed.git_version_info import version as ds_version from .simple_model import SimpleModel, SimpleOptimizer, random_dataloader, args_from_dict +import os base_ds_config = { "elasticity": { @@ -107,6 +108,42 @@ def test_empty_config(): target_deepspeed_version=ds_version) +def test_model_parallel_v1_invalid(): + ds_config = base_ds_config.copy() + ds_config["elasticity"]["model_parallel_size"] = 4 + ds_config["elasticity"]["num_gpus_per_node"] = 8 + ds_config["elasticity"]["version"] = 0.1 + + with pytest.raises(deepspeed.elasticity.config.ElasticityError): + deepspeed.elasticity.compute_elastic_config(ds_config=ds_config, + target_deepspeed_version=ds_version) + + +def test_model_parallel_v2_invalid(): + ds_config = base_ds_config.copy() + ds_config["elasticity"]["model_parallel_size"] = 16 + ds_config["elasticity"]["num_gpus_per_node"] = 8 + ds_config["elasticity"]["version"] = 0.2 + + with pytest.raises(deepspeed.elasticity.config.ElasticityError): + deepspeed.elasticity.compute_elastic_config(ds_config=ds_config, + target_deepspeed_version=ds_version, + world_size=16) + + +def test_model_parallel_v2_valid(): + + ds_config = base_ds_config.copy() + ds_config["elasticity"]["model_parallel_size"] = 4 + ds_config["elasticity"]["num_gpus_per_node"] = 8 + ds_config["elasticity"]["version"] = 0.2 + + os.environ["WORLD_SIZE"] = str(16) + deepspeed.elasticity.compute_elastic_config(ds_config=ds_config, + target_deepspeed_version=ds_version) + os.environ.pop("WORLD_SIZE") + + @pytest.mark.parametrize('key, value', [('micro_batch_sizes', [1, From 6ed806600e6c2be6ec17a19fc43db0eb29817ff3 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 28 Jul 2022 22:04:15 +0000 Subject: [PATCH 22/27] fixed a typo --- deepspeed/elasticity/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/deepspeed/elasticity/config.py b/deepspeed/elasticity/config.py index ea03a7c8a4ad..ffbce7028e03 100644 --- a/deepspeed/elasticity/config.py +++ b/deepspeed/elasticity/config.py @@ -100,7 +100,7 @@ def __init__(self, param_dict): NUM_GPUS_PER_NODE_DEFAULT) if self.num_gpus_per_node < 1: raise ElasticityConfigError( - "NUmber of GPUs per node cannot be less than 1, " + "Number of GPUs per node cannot be less than 1, " f"given number of GPUs per node: {self.num_gpus_per_node}") self.min_time = param_dict.get(MIN_TIME, MIN_TIME_DEFAULT) From f2405bd317c97cd3bcbfa1bf08283dde6846dc21 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 28 Jul 2022 22:14:21 +0000 Subject: [PATCH 23/27] fixed test_elastic --- tests/unit/test_elastic.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/unit/test_elastic.py b/tests/unit/test_elastic.py index dc6d43e8a75e..4ed2c0dd0c95 100644 --- a/tests/unit/test_elastic.py +++ b/tests/unit/test_elastic.py @@ -4,7 +4,6 @@ from deepspeed.git_version_info import version as ds_version import os from .simple_model import SimpleModel, args_from_dict ->>>>>>> staging-ft-elastic-v1 base_ds_config = { "elasticity": { From 66205a1246e97b7c3b3cc22fca72b6ac0d396950 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Thu, 28 Jul 2022 22:25:29 +0000 Subject: [PATCH 24/27] removed extra imports --- deepspeed/elasticity/elastic_agent.py | 8 ++++---- deepspeed/elasticity/elasticity.py | 2 +- deepspeed/launcher/launch.py | 4 ++-- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/deepspeed/elasticity/elastic_agent.py b/deepspeed/elasticity/elastic_agent.py index e9ac126b9a84..8d0981612725 100644 --- a/deepspeed/elasticity/elastic_agent.py +++ b/deepspeed/elasticity/elastic_agent.py @@ -1,8 +1,8 @@ from torch.distributed.elastic.agent.server.local_elastic_agent import LocalElasticAgent from typing import Any, Dict, Optional, Tuple -from datetime import datetime, timedelta -from torch.distributed.elastic.agent.server.api import RunResult, log, WorkerState, _get_socket_with_port -from torch.distributed.elastic.metrics import prof, put_metric +from datetime import datetime +from torch.distributed.elastic.agent.server.api import log, _get_socket_with_port +from torch.distributed.elastic.metrics import put_metric from torch.distributed.elastic.agent.server.api import ( RunResult, WorkerGroup, @@ -12,7 +12,7 @@ from torch.distributed import Store import time import os -from torch.distributed.elastic.multiprocessing import PContext, start_processes +from torch.distributed.elastic.multiprocessing import start_processes from torch.distributed.elastic.utils import macros import shutil import copy diff --git a/deepspeed/elasticity/elasticity.py b/deepspeed/elasticity/elasticity.py index ff58f78b0974..17a8b6ecf394 100644 --- a/deepspeed/elasticity/elasticity.py +++ b/deepspeed/elasticity/elasticity.py @@ -13,7 +13,7 @@ MINIMUM_DEEPSPEED_VERSION, DEEPSPEED_ELASTICITY_CONFIG from ..git_version_info import version as __version__ from ..utils import logger -import subprocess + # Thirty eight smallest highly composite numbers. The list should # be enough to support up to 720K batch size. HCN_LIST = [ diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 967eb645ead8..d59a7ab36ce3 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -17,6 +17,7 @@ import signal import psutil from collections import defaultdict +from typing import Dict from argparse import ArgumentParser, REMAINDER from ..constants import TORCH_DISTRIBUTED_DEFAULT_PORT from ..nebula.constants import DLTS_POD_ENV_PATH @@ -177,8 +178,7 @@ def main(): if not is_torch_elastic_compatible(): if args.enable_elastic_training: logger.info(f"Disabling elastic training support as \ - PyTorch version should be greater than 1.11.x.\ - Current version is {torch.__version__}") + PyTorch version should be greater than 1.11.x") args.enable_elastic_training = False if os.path.exists(DLTS_POD_ENV_PATH): From 7e601b3a3f1ca9e1c801119b264dfaf60b3e83be Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 29 Jul 2022 15:31:33 +0000 Subject: [PATCH 25/27] renamed min and max nodes arguments --- deepspeed/launcher/launch.py | 18 +++++++++--------- deepspeed/launcher/multinode_runner.py | 4 ++-- deepspeed/launcher/runner.py | 8 ++++---- 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index d59a7ab36ce3..981e05742388 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -71,12 +71,12 @@ def parse_args(): action="store_true", help="Enable elastic training support.") - parser.add_argument("--min_nodes", + parser.add_argument("--min_elastic_nodes", type=int, default=-1, help="Min number of nodes in elastic training.") - parser.add_argument("--max_nodes", + parser.add_argument("--max_elastic_nodes", type=int, default=-1, help="Max number of nodes in elastic training.") @@ -227,11 +227,11 @@ def main(): import torch.distributed.elastic.rendezvous.registry as rdzv_registry from torch.distributed.elastic.multiprocessing import Std - if args.min_nodes == -1: - args.min_nodes = 1 - if args.max_nodes == -1: - args.max_nodes = args.nnodes - assert args.max_nodes > 0 and args.min_nodes > 0 , "Max and Min nodes should be positive" + if args.min_elastic_nodes == -1: + args.min_elastic_nodes = 1 + if args.max_elastic_nodes == -1: + args.max_elastic_nodes = args.nnodes + assert args.max_elastic_nodes > 0 and args.min_elastic_nodes > 0 , "Max and Min nodes should be positive" current_env["NCCL_ASYNC_ERROR_HANDLING"] = str(1) @@ -254,8 +254,8 @@ def main(): endpoint=args.master_addr + ":" + str(args.master_port), run_id='123456789', - min_nodes=args.min_nodes, - max_nodes=args.max_nodes, + min_nodes=args.min_elastic_nodes, + max_nodes=args.max_elastic_nodes, **rdzv_configs) spec = WorkerSpec( diff --git a/deepspeed/launcher/multinode_runner.py b/deepspeed/launcher/multinode_runner.py index 9ff7e644766c..7c2828e75fc7 100644 --- a/deepspeed/launcher/multinode_runner.py +++ b/deepspeed/launcher/multinode_runner.py @@ -96,8 +96,8 @@ def get_cmd(self, environment, active_resources): deepspeed_launch += ["--save_pid", f"{os.getpid()}"] if self.args.elastic_training: deepspeed_launch.append("--enable_elastic_training") - deepspeed_launch.append(f"--max_nodes={self.args.max_num_nodes}") - deepspeed_launch.append(f"--min_nodes={self.args.min_num_nodes}") + deepspeed_launch.append(f"--max_elastic_nodes={self.args.max_elastic_nodes}") + deepspeed_launch.append(f"--min_elastic_nodes={self.args.min_elastic_nodes}") cmd_to_search = [i + "\\" for i in deepspeed_launch[2:6]] diff --git a/deepspeed/launcher/runner.py b/deepspeed/launcher/runner.py index 4d1c0821cfe3..f35fb3994ac4 100755 --- a/deepspeed/launcher/runner.py +++ b/deepspeed/launcher/runner.py @@ -78,13 +78,13 @@ def parse_args(args=None): help="Total number of worker nodes to run on, this will use " "the top N hosts from the given hostfile.") - parser.add_argument("--min_num_nodes", + parser.add_argument("--min_elastic_nodes", type=int, default=-1, help="Minimum number of nodes to run elastic training on. " "Default is 1 when elastic training is enabled") - parser.add_argument("--max_num_nodes", + parser.add_argument("--max_elastic_nodes", type=int, default=-1, help="Maximum number of nodes to run elastic training on. " @@ -460,8 +460,8 @@ def main(args=None): deepspeed_launch += ["--save_pid", f"{os.getpid()}"] if args.elastic_training: deepspeed_launch.append("--enable_elastic_training") - deepspeed_launch.append(f"--max_nodes={args.max_num_nodes}") - deepspeed_launch.append(f"--min_nodes={args.min_num_nodes}") + deepspeed_launch.append(f"--max_elastic_nodes={args.max_elastic_nodes}") + deepspeed_launch.append(f"--min_elastic_nodes={args.min_elastic_nodes}") cmd = deepspeed_launch + [args.user_script] + args.user_args else: args.launcher = args.launcher.lower() From 6de2cd87da7aa2285ec2280e81a4a465c077dff3 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 29 Jul 2022 15:48:17 +0000 Subject: [PATCH 26/27] use deafult elastic ID --- deepspeed/launcher/launch.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index 981e05742388..c9064d75018d 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -23,6 +23,7 @@ from ..nebula.constants import DLTS_POD_ENV_PATH from ..utils import logger from ..elasticity import is_torch_elastic_compatible +from .constants import ELASTIC_TRAINING_ID_DEFAULT PID_FILE_BASEPATH = "/tmp" @@ -253,7 +254,7 @@ def main(): rdzv_parameters = RendezvousParameters(backend='c10d', endpoint=args.master_addr + ":" + str(args.master_port), - run_id='123456789', + run_id=ELASTIC_TRAINING_ID_DEFAULT, min_nodes=args.min_elastic_nodes, max_nodes=args.max_elastic_nodes, **rdzv_configs) From ae26b52ae49eb4fdae740a01e809878ba59e6985 Mon Sep 17 00:00:00 2001 From: Arpan Jain Date: Fri, 29 Jul 2022 15:59:42 +0000 Subject: [PATCH 27/27] expose elastic run id as an env variable --- deepspeed/launcher/launch.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/deepspeed/launcher/launch.py b/deepspeed/launcher/launch.py index c9064d75018d..f39530552055 100755 --- a/deepspeed/launcher/launch.py +++ b/deepspeed/launcher/launch.py @@ -251,10 +251,13 @@ def main(): cmd_args = cmd[1:] rdzv_configs: Dict[str, str] = {'timeout': 100} + run_id = os.environ.get("ELASTIC_RUN_ID", ELASTIC_TRAINING_ID_DEFAULT) + + # Creating config for rendezvous class rdzv_parameters = RendezvousParameters(backend='c10d', endpoint=args.master_addr + ":" + str(args.master_port), - run_id=ELASTIC_TRAINING_ID_DEFAULT, + run_id=run_id, min_nodes=args.min_elastic_nodes, max_nodes=args.max_elastic_nodes, **rdzv_configs)