[Bugfix] Give each data-parallel engine its own global RNG streams (#59788)
Signed-off-by: aoshen02 <aoshen@inferact.ai> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
877ddcc456
commit
d09ff77637
@@ -14,6 +14,7 @@ from vllm.utils.torch_utils import (
|
||||
is_lossless_cast,
|
||||
is_quantized_kv_cache,
|
||||
set_default_torch_dtype,
|
||||
set_random_seed,
|
||||
set_torch_threads_for_runtime,
|
||||
startup_omp_num_threads,
|
||||
)
|
||||
@@ -217,3 +218,17 @@ def test_async_tensor_h2d_staging(device):
|
||||
async_tensor_h2d([1, 2, 3], device=device, dtype=torch.int32),
|
||||
torch.tensor([1, 2, 3], dtype=torch.int32, device=device),
|
||||
)
|
||||
|
||||
|
||||
def test_set_random_seed_differs_per_data_parallel_index():
|
||||
"""Unseeded requests draw from the global RNGs; DP engines sharing the engine
|
||||
seed must get distinct, reproducible streams, and index 0 keeps the seed."""
|
||||
|
||||
def draws(*args):
|
||||
set_random_seed(*args)
|
||||
return np.random.randint(2**31, size=4).tolist(), torch.rand(4).tolist()
|
||||
|
||||
assert draws(42, 0) == draws(42)
|
||||
assert draws(42, 1) == draws(42, 1)
|
||||
assert len({str(draws(42, i)) for i in range(4)}) == 4
|
||||
assert draws(42, 1) != draws(43, 1)
|
||||
|
||||
@@ -530,8 +530,12 @@ def kv_cache_dtype_str_to_dtype(
|
||||
return STR_DTYPE_TO_TORCH_DTYPE[kv_cache_dtype]
|
||||
|
||||
|
||||
def set_random_seed(seed: int | None) -> None:
|
||||
def set_random_seed(seed: int | None, data_parallel_index: int = 0) -> None:
|
||||
if seed is not None:
|
||||
if data_parallel_index:
|
||||
# DP engines share the seed but must not sample unseeded requests alike.
|
||||
ss = np.random.SeedSequence((seed, data_parallel_index))
|
||||
seed = int(ss.generate_state(1)[0])
|
||||
random.seed(seed)
|
||||
np.random.seed(seed)
|
||||
torch.manual_seed(seed)
|
||||
|
||||
@@ -268,7 +268,9 @@ class CPUWorker(Worker):
|
||||
self.model_runner.warming_up_model()
|
||||
# Reset the seed to ensure that the random state is not affected by
|
||||
# the model initialization and profiling.
|
||||
set_random_seed(self.model_config.seed)
|
||||
set_random_seed(
|
||||
self.model_config.seed, self.parallel_config.data_parallel_index
|
||||
)
|
||||
return CompilationTimes(
|
||||
language_model=self.compilation_config.compilation_time,
|
||||
encoder=self.compilation_config.encoder_compilation_time,
|
||||
|
||||
@@ -1020,7 +1020,9 @@ class Worker(WorkerBase):
|
||||
|
||||
# Reset the seed to ensure that the random state is not affected by
|
||||
# the model initialization and profiling.
|
||||
set_random_seed(self.model_config.seed)
|
||||
set_random_seed(
|
||||
self.model_config.seed, self.parallel_config.data_parallel_index
|
||||
)
|
||||
|
||||
# Eagerly trigger inductor's once-per-process lazy inits during
|
||||
# warmup (rather than on a later compile cache-miss at runtime).
|
||||
|
||||
Reference in New Issue
Block a user