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