fix: add zero-guards for divisor/scale edge cases (deepspeedai#7838)

Prevent ZeroDivisionError and silent non-finite propagation at four
reported sites:

1. groups._ensure_divisibility: reject denominator == 0 before modulo
2. ThroughputTimer: reject non-positive steps_per_output at init and
   report-boundary time
3. inference_utils.ceil_div: reject divisor == 0
4. HPU FPQuantizer.dequantize: reject zero/non-finite scales before
   1/scale

Adds focused unit tests covering each failure mode.

Picks up the inactive draft deepspeedai#7855.

Fixes deepspeedai#7838

Signed-off-by: liuyun7345 <liuyun7345@sina.com>

Author: liuyun7345

deepspeed/inference/v2/inference_utils.py

@@ -102,4 +102,6 @@ def ceil_div(a: int, b: int) -> int:
     """
     Return ceil(a / b).
     """
+    if b == 0:
+        raise ValueError(f"ceil_div divisor must be non-zero (got a={a}, b={b})")
     return -(-a // b)

deepspeed/utils/groups.py

@@ -63,6 +63,7 @@ def initialize(ep_size=1, mpu=None):
 
 def _ensure_divisibility(numerator, denominator):
     """Ensure that numerator is divisible by the denominator."""
+    assert denominator != 0, f'denominator must be non-zero (got numerator={numerator}, denominator={denominator})'
     assert numerator % denominator == 0, '{} is not divisible by {}'.format(numerator, denominator)
 
 

deepspeed/utils/timer.py

@@ -211,6 +211,8 @@ def __init__(self, config, batch_size, start_step=2, steps_per_output=None, moni
         self.global_step_count = 0
         self.total_elapsed_time = 0
         self.step_elapsed_time = 0
+        if steps_per_output is not None and steps_per_output <= 0:
+            raise ValueError(f"steps_per_output must be a positive integer or None, got {steps_per_output}")
         self.steps_per_output = steps_per_output
         self.monitor_memory = monitor_memory
         self.logging = logging_fn
@@ -241,6 +243,9 @@ def start(self):
     def _is_report_boundary(self):
         if self.steps_per_output is None:
             return False
+        # Guard against mutation to 0 after construction (see #7838).
+        if self.steps_per_output <= 0:
+            raise ValueError(f"steps_per_output must be a positive integer, got {self.steps_per_output}")
         return self.global_step_count % self.steps_per_output == 0
 
     def stop(self, global_step=False, report_speed=True):

op_builder/hpu/fp_quantizer.py

@@ -4,6 +4,8 @@
 
 # DeepSpeed Team
 
+import math
+
 import torch
 try:
     # is op_builder from deepspeed or a 3p version? this should only succeed if it's deepspeed
@@ -54,6 +56,15 @@ def selective_dequantize(cls, val_q, scales, indexes, group_size, q_mantisa_bits
 
     @classmethod
     def dequantize(cls, fp_out, input_q, scale, group_size, q_mantisa_bits, q_exponent_bits):
+        # Reject zero / non-finite scales before inverse-scale computation to avoid
+        # silently propagating inf/nan into dequantized outputs (#7838).
+        if torch.is_tensor(scale):
+            if (not torch.isfinite(scale).all()) or (scale == 0).any():
+                raise ValueError("FPQuantizer.dequantize requires finite non-zero scale values")
+        else:
+            scale_f = float(scale)
+            if scale_f == 0.0 or not math.isfinite(scale_f):
+                raise ValueError("FPQuantizer.dequantize requires a finite non-zero scale")
         orig_shape = fp_out.shape
         orig_dtype = fp_out.dtype
         dequant_out = torch.ops.hpu.cast_from_fp8(input_q, (1.0 / scale), orig_dtype).view(orig_shape)

tests/unit/utils/test_zero_guards.py

@@ -0,0 +1,97 @@
+# Copyright (c) Microsoft Corporation.
+# SPDX-License-Identifier: Apache-2.0
+
+# DeepSpeed Team
+
+"""Regression tests for zero/division safety gaps reported in #7838."""
+
+import math
+
+import pytest
+import torch
+
+from deepspeed.utils.groups import _ensure_divisibility
+from deepspeed.utils.timer import ThroughputTimer
+from deepspeed.inference.v2.inference_utils import ceil_div
+
+
+class _DummyTimerConfig:
+    enabled = False
+    synchronized = False
+
+
+def test_ensure_divisibility_rejects_zero_denominator():
+    with pytest.raises(AssertionError, match="non-zero"):
+        _ensure_divisibility(8, 0)
+
+
+def test_ensure_divisibility_accepts_valid_inputs():
+    _ensure_divisibility(8, 2)
+    _ensure_divisibility(0, 4)
+
+
+def test_ceil_div_rejects_zero_divisor():
+    with pytest.raises(ValueError, match="non-zero"):
+        ceil_div(10, 0)
+
+
+def test_ceil_div_matches_math_ceil():
+    assert ceil_div(10, 3) == math.ceil(10 / 3)
+    assert ceil_div(9, 3) == 3
+    assert ceil_div(1, 1) == 1
+
+
+def test_throughput_timer_rejects_zero_steps_per_output():
+    with pytest.raises(ValueError, match="positive"):
+        ThroughputTimer(_DummyTimerConfig(), batch_size=1, steps_per_output=0)
+
+
+def test_throughput_timer_rejects_negative_steps_per_output():
+    with pytest.raises(ValueError, match="positive"):
+        ThroughputTimer(_DummyTimerConfig(), batch_size=1, steps_per_output=-1)
+
+
+def test_throughput_timer_report_boundary_guards_mutated_zero():
+    timer = ThroughputTimer(_DummyTimerConfig(), batch_size=1, steps_per_output=2)
+    timer.steps_per_output = 0
+    with pytest.raises(ValueError, match="positive"):
+        timer._is_report_boundary()
+
+
+def test_throughput_timer_report_boundary_none_is_safe():
+    timer = ThroughputTimer(_DummyTimerConfig(), batch_size=1, steps_per_output=None)
+    assert timer._is_report_boundary() is False
+
+
+def _import_hpu_fp_quantizer_builder():
+    try:
+        from op_builder.hpu.fp_quantizer import FPQuantizerBuilder
+        return FPQuantizerBuilder
+    except ImportError:
+        pytest.skip("HPU FPQuantizer builder is not available")
+
+
+def test_hpu_fp_quantizer_dequantize_rejects_zero_scale():
+    FPQuantizerBuilder = _import_hpu_fp_quantizer_builder()
+    scale = torch.tensor([0.0, 1.0])
+    fp_out = torch.empty(2, 4)
+    input_q = torch.empty(2, 4)
+    with pytest.raises(ValueError, match="finite non-zero"):
+        FPQuantizerBuilder.dequantize(fp_out, input_q, scale, group_size=4, q_mantisa_bits=3, q_exponent_bits=4)
+
+
+def test_hpu_fp_quantizer_dequantize_rejects_nonfinite_scale():
+    FPQuantizerBuilder = _import_hpu_fp_quantizer_builder()
+    scale = torch.tensor([float("nan"), 1.0])
+    fp_out = torch.empty(2, 4)
+    input_q = torch.empty(2, 4)
+    with pytest.raises(ValueError, match="finite non-zero"):
+        FPQuantizerBuilder.dequantize(fp_out, input_q, scale, group_size=4, q_mantisa_bits=3, q_exponent_bits=4)
+
+
+def test_hpu_fp_quantizer_dequantize_rejects_zero_scalar_scale():
+    FPQuantizerBuilder = _import_hpu_fp_quantizer_builder()
+    fp_out = torch.empty(2, 4)
+    input_q = torch.empty(2, 4)
+    with pytest.raises(ValueError, match="finite non-zero"):
+        FPQuantizerBuilder.dequantize(fp_out, input_q, 0.0, group_size=4, q_mantisa_bits=3, q_exponent_bits=4)