DeepSpeed content for 1.22.0

Signed-off-by: SW publisher <[email protected]>

Author: SW publisher

.pre-commit-config.yaml

@@ -5,52 +5,6 @@
 # Learn more about CODEOWNERS syntax here:
 # https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners
 
+*          [email protected] [email protected] [email protected] [email protected] [email protected] [email protected]
 
-# top-level repo folders
-/.github/ @loadams
-/azure/ @awan-10
-/benchmarks/ @awan-10 @tjruwase
-/bin/ @loadams
-/csrc/ @awan-10
-/deepspeed/ @loadams @tjruwase
-/docker/ @awan-10
-/docs/ @loadams @tjruwase
-/examples/ @awan-10 @tohtana
-/op_builder/ @loadams @tjruwase @jomayeri
-/release/ @loadams
-/requirements/ @loadams
-/scripts/ @awan-10
-/tests/ @tjruwase @loadams @tohtana
-
-# deepspeed
-/deepspeed/autotuning/ @loadams
-/deepspeed/checkpoint/ @tjruwase
-/deepspeed/comm/ @awan-10
-/deepspeed/compression/ @tjruwase
-/deepspeed/elasticity/ @awan-10
-/deepspeed/launcher/ @loadams
-/deepspeed/module_inject/ @awan-10
-/deepspeed/moe/ @tohtana
-/deepspeed/monitor/ @awan-10
-/deepspeed/nebula/ @tjruwase
-/deepspeed/ops/ @tohtana
-/deepspeed/pipe/ @tohtana @loadams
-/deepspeed/profiling/ @loadams
-/deepspeed/utils/ @tjruwase @awan-10
-
-# inference
-/deepspeed/inference/ @awan-10
-/deepspeed/model_implementations/ @awan-10
-
-# training
-/deepspeed/runtime/ @tjruwase @tohtana
-/deepspeed/runtime/activation_checkpointing/ @tjruwase
-/deepspeed/runtime/checkpoint_engine/ @tjruwase
-/deepspeed/runtime/comm/ @awan-10
-/deepspeed/runtime/compression/ @awan-10
-/deepspeed/runtime/data_pipeline/ @tjruwase
-/deepspeed/runtime/fp16/ @tjruwase
-/deepspeed/runtime/fp16/onebit/ @awan-10
-/deepspeed/runtime/pipe/ @loadams
-/deepspeed/runtime/swap_tensor/ @tjruwase
-/deepspeed/runtime/zero/ @tjruwase
+CODEOWNERS [email protected] [email protected]

CODEOWNERS

@@ -21,8 +21,11 @@ def __init__(self):
         self.apply_hpu_workarounds()
         try:
             import habana_frameworks.torch.hpu as hpu
-            hpu.setDeterministic(True)
             self.hpu = hpu
+            torch.use_deterministic_algorithms(True)
+            # TODO: [SW-215614] remove this WA when SW-208658 is resolved.
+            torch.utils.deterministic.fill_uninitialized_memory = False
+
         except ImportError as e:
             raise ValueError(
                 f"HPU_Accelerator requires habana_frameworks.torch.hpu, which is not installed on this system.")
@@ -299,6 +302,14 @@ def get_op_builder(self, class_name):
         else:
             return self.class_dict['NotImplementedBuilder'] if 'NotImplementedBuilder' in self.class_dict else None
 
+    #shall be removed once moving to torch.compile
+    def wrap_in_hpu_graph(self, module):
+        if self.hpu.is_lazy():
+            module = self.hpu.wrap_in_hpu_graph(module)
+        else:
+            print("Warning: hpu graphs in eager mode is not supported, ignoring")
+        return module
+
     def build_extension(self):
         from torch.utils.cpp_extension import BuildExtension
         return BuildExtension
@@ -307,6 +318,7 @@ def export_envs(self):
         return []
 
     def visible_devices_envs(self):
+        # TODO SW-195658: remove WA to not return HABANA_VISIBLE_MODULES once SW-195657 is resolved
         # Current way deepspeed set this env var is not applicable with all HPU instances
         # User has to follow instructions in:
         # https://docs.habana.ai/en/latest/PyTorch/Reference/PT_Multiple_Tenants_on_HPU/Multiple_Workloads_Single_Docker.html

accelerator/hpu_accelerator.py

@@ -67,7 +67,7 @@ def get_accelerator():
                     f"XPU_Accelerator requires intel_extension_for_pytorch, which is not installed on this system.")
         elif accelerator_name == "xpu.external":
             try:
-                import intel_extension_for_deepspeed  # noqa: F401 # type: ignore
+                from intel_extension_for_deepspeed import XPU_Accelerator  # noqa: F401 # type: ignore
             except ImportError as e:
                 raise ValueError(
                     f"XPU_Accelerator external requires intel_extension_for_deepspeed, which is not installed on this system."
@@ -193,6 +193,12 @@ def get_accelerator():
         ds_accelerator = CPU_Accelerator()
     elif accelerator_name == "xpu.external":
         # XPU_Accelerator is already imported in detection stage
+        try:
+            from intel_extension_for_deepspeed import XPU_Accelerator  # noqa: F811
+        except ImportError as e:
+            raise ValueError(
+                f"XPU_Accelerator external requires intel_extension_for_deepspeed, which is not installed on this system."
+            )
         ds_accelerator = XPU_Accelerator()
     elif accelerator_name == "xpu":
         from .xpu_accelerator import XPU_Accelerator
@@ -223,7 +229,7 @@ def get_accelerator():
 def set_accelerator(accel_obj):
     global ds_accelerator
     _validate_accelerator(accel_obj)
-    if accel_logger is not None:
+    if accel_logger is not None and accel_obj is not None:
         accel_logger.info(f"Setting ds_accelerator to {accel_obj._name} (model specified)")
     ds_accelerator = accel_obj
 

accelerator/real_accelerator.py

@@ -0,0 +1 @@
++hpu.synapse.v1.22.0

build.txt

@@ -24,6 +24,7 @@
 
 at::Tensor quantize(torch::Tensor& out,
                     torch::Tensor& val,
+                    torch::Tensor& scale,
                     int group_size,
                     int stochastic_rounding,
                     int q_bits,
@@ -59,6 +60,7 @@ at::Tensor quantize(torch::Tensor& out,
 
 void dequantize(torch::Tensor& val,
                 torch::Tensor& val_q,
+                torch::Tensor& scale,
                 int group_size,
                 int q_mantisa_bits,
                 int q_exponent_bits)

csrc/fp_quantizer/fp_quantize.cpp

@@ -81,7 +81,7 @@ def __init__(self, args, active_resources):
         if not os.path.exists(self.results_dir):
             try:
                 os.makedirs(self.results_dir, exist_ok=True)
-                logger.info(f"Created autotuning results directory: {self.exps_dir}")
+                logger.info(f"Created autotuning results directory: {self.results_dir}")
             except:
                 logger.error(
                     f"Failed to create {self.results_dir}, please check `results_dir` in the autotuning config file is accessible by all the nodes in the job."

deepspeed/autotuning/autotuner.py

@@ -144,7 +144,7 @@
     "zero_optimization": {
         "stage": 3
     },
-    "memory_break_down": False
+    "memory_breakdown": False
 }
 
 DEFAULT_TUNING_SPACE_ZERO_0 = {"zero_optimization": {"stage": 0}}

deepspeed/autotuning/constants.py

@@ -77,27 +77,12 @@ def run_collective(self, name, **kwargs):
             return CCLHandler(self.ccl_comm_op)
 
     def all_reduce(self, tensor, op=ReduceOp.SUM, group=None, async_op=False):
-        use_caching = False
-        if use_caching:
-            match_id = f"{tensor.size()}-{op}"
-            name = "all_reduce_caching"
-            if name in self.available_coll:
-                group = self.get_all_ranks_from_group(group)
-                return self.ccl_comm_op.all_reduce_caching(tensor, op, match_id, group, async_op)
-            else:
-                return self.run_collective(name=name,
-                                           tensor=tensor,
-                                           op=op,
-                                           match_id=match_id,
-                                           group=group,
-                                           async_op=async_op)
+        name = "all_reduce"
+        if name in self.available_coll:
+            group = self.get_all_ranks_from_group(group)
+            return self.ccl_comm_op.all_reduce(tensor, op, group, async_op)
         else:
-            name = "all_reduce"
-            if name in self.available_coll:
-                group = self.get_all_ranks_from_group(group)
-                return self.ccl_comm_op.all_reduce(tensor, op, group, async_op)
-            else:
-                return self.run_collective(name=name, tensor=tensor, op=op, group=group, async_op=async_op)
+            return self.run_collective(name=name, tensor=tensor, op=op, group=group, async_op=async_op)
 
     def inference_all_reduce(self, tensor, op=ReduceOp.SUM, group=None):
         name = "inference_all_reduce"