(self, config: Union[str, dict], mpu=None, mesh_device=None)
| 680 | class DeepSpeedConfig(object): |
| 681 | |
| 682 | def __init__(self, config: Union[str, dict], mpu=None, mesh_device=None): |
| 683 | super(DeepSpeedConfig, self).__init__() |
| 684 | if isinstance(config, dict): |
| 685 | self._param_dict = config |
| 686 | elif os.path.exists(config): |
| 687 | self._param_dict = hjson.load(open(config, "r"), object_pairs_hook=dict_raise_error_on_duplicate_keys) |
| 688 | else: |
| 689 | try: |
| 690 | config_decoded = base64.urlsafe_b64decode(config).decode('utf-8') |
| 691 | self._param_dict = hjson.loads(config_decoded) |
| 692 | except (UnicodeDecodeError, AttributeError): |
| 693 | raise ValueError( |
| 694 | f"Expected a string path to an existing deepspeed config, or a dictionary or a valid base64. Received: {config}" |
| 695 | ) |
| 696 | |
| 697 | try: |
| 698 | self.global_rank = dist.get_rank() |
| 699 | if mpu is not None: |
| 700 | # Ulysses SP |
| 701 | if not hasattr(mpu, "get_data_parallel_world_size"): |
| 702 | self.world_size = dist.get_world_size() / mpu.get_sequence_parallel_world_size() |
| 703 | else: |
| 704 | self.world_size = mpu.get_data_parallel_world_size() |
| 705 | elif mesh_device is not None: |
| 706 | self.world_size = dist.get_world_size(mesh_device.get_group(mesh_dim="data_parallel")) |
| 707 | else: |
| 708 | # HF zero.init case where there is no mpu |
| 709 | if "sequence_parallel_size" in config: |
| 710 | self.world_size = dist.get_world_size() / config["sequence_parallel_size"] |
| 711 | else: |
| 712 | self.world_size = dist.get_world_size() |
| 713 | except (RuntimeError, AssertionError, AttributeError): |
| 714 | self.global_rank = 0 |
| 715 | self.world_size = 1 |
| 716 | logger.info(f"Config mesh_device {mesh_device} world_size = {self.world_size}") |
| 717 | # If elastic-mode enabled, update compute + update _param_dict |
| 718 | self.elasticity_enabled = elasticity_enabled(self._param_dict) |
| 719 | if self.elasticity_enabled: |
| 720 | logger.info("DeepSpeed elasticity support enabled") |
| 721 | final_batch_size, valid_gpus, micro_batch_size = compute_elastic_config( |
| 722 | ds_config=self._param_dict, |
| 723 | target_deepspeed_version=__version__, |
| 724 | world_size=self.world_size, |
| 725 | ) |
| 726 | |
| 727 | elastic_dict = self._param_dict[ELASTICITY] |
| 728 | |
| 729 | # Ensure the resource scheduler saw the same elastic config we are using at runtime |
| 730 | ensure_immutable_elastic_config(runtime_elastic_config_dict=elastic_dict) |
| 731 | |
| 732 | self.elastic_model_parallel_size = elastic_dict.get(MODEL_PARALLEL_SIZE, MODEL_PARALLEL_SIZE_DEFAULT) |
| 733 | if self.elastic_model_parallel_size < 1: |
| 734 | raise ElasticityConfigError("Model-Parallel size cannot be less than 1, " |
| 735 | f"given model-parallel size: {self.elastic_model_parallel_size}") |
| 736 | |
| 737 | self.num_gpus_per_node = elastic_dict.get(NUM_GPUS_PER_NODE, NUM_GPUS_PER_NODE_DEFAULT) |
| 738 | if self.num_gpus_per_node < 1: |
| 739 | raise ElasticityConfigError("NUmber of GPUs per node cannot be less than 1, " |
nothing calls this directly
no test coverage detected