(self, params, has_moe_params=None)
| 283 | return False |
| 284 | |
| 285 | def has_overflow(self, params, has_moe_params=None): |
| 286 | if has_moe_params is None: |
| 287 | has_moe_params = self.has_moe_params |
| 288 | overflow = self.has_overflow_serial(params) |
| 289 | # Since each model parallel GPU carries only part of the model, |
| 290 | # make sure overflow flag is synced across all the model parallel GPUs |
| 291 | overflow_gpu = get_accelerator().ByteTensor([overflow]) |
| 292 | # deepspeed.comm.all_reduce(overflow_gpu, |
| 293 | # op=deepspeed.comm.ReduceOp.MAX, |
| 294 | # group=mpu.get_model_parallel_group()) |
| 295 | if has_moe_params: |
| 296 | # All reduce this across expert_parallel_group, so that if an expert |
| 297 | # overflows, we detect it here |
| 298 | dist.all_reduce(overflow_gpu, op=dist.ReduceOp.MAX, group=groups._get_max_expert_parallel_group()) |
| 299 | if self.zero_reduce_scatter: |
| 300 | dist.all_reduce(overflow_gpu, op=dist.ReduceOp.MAX, group=dist.get_world_group()) |
| 301 | elif self.mpu is not None: |
| 302 | if self.deepspeed is not None: |
| 303 | using_pipeline = hasattr(self.deepspeed, 'pipeline_enable_backward_allreduce') |
| 304 | if (using_pipeline and self.deepspeed.pipeline_enable_backward_allreduce |
| 305 | is False) or (not using_pipeline and self.deepspeed.enable_backward_allreduce is False): |
| 306 | dist.all_reduce(overflow_gpu, op=dist.ReduceOp.MAX, group=self.mpu.get_data_parallel_group()) |
| 307 | dist.all_reduce(overflow_gpu, op=dist.ReduceOp.MAX, group=self.mpu.get_model_parallel_group()) |
| 308 | elif self.deepspeed is not None and self.deepspeed.enable_backward_allreduce is False: |
| 309 | dist.all_reduce(overflow_gpu, op=dist.ReduceOp.MAX, group=dist.get_world_group()) |
| 310 | |
| 311 | overflow = overflow_gpu[0].item() |
| 312 | return bool(overflow) |
| 313 | |
| 314 | # `x` is a torch.Tensor |
| 315 | @staticmethod |
no test coverage detected