Skip to content

[Intel HPU] fix memory fragmentation issue and fix moe all_reduce issue #5357

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
EmmonsCurse merged 1 commit into from
Dec 4, 2025
Merged

[Intel HPU] fix memory fragmentation issue and fix moe all_reduce issue #5357

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 0 additions & 6 deletions fastdeploy/model_executor/layers/backends/intel_hpu/moe/fused_moe_hpu_backend.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
from paddle import nn

from fastdeploy import envs
from fastdeploy.distributed.communication import tensor_model_parallel_all_reduce_custom
from fastdeploy.model_executor.layers.moe.fused_moe_backend_base import (
UnquantizedFusedMoEMethod,
)
Expand Down Expand Up @@ -96,8 +95,6 @@ def apply_tp(
experts_max=layer.expert_id_offset + layer.num_local_experts - 1,
chunk_size=chunk_size,
)
if layer.reduce_results and layer.tp_size > 1:
tensor_model_parallel_all_reduce_custom(fused_moe_out)

return fused_moe_out

Expand Down Expand Up @@ -198,7 +195,4 @@ def apply_tp(
chunk_size=chunk_size,
)

if layer.reduce_results and layer.tp_size > 1:
tensor_model_parallel_all_reduce_custom(fused_moe_out)

return fused_moe_out
10 changes: 8 additions & 2 deletions fastdeploy/model_executor/layers/moe/moe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@
from paddleformers.utils.log import logger

from fastdeploy import envs
from fastdeploy.distributed.communication import tensor_model_parallel_all_reduce
from fastdeploy.distributed.communication import (
tensor_model_parallel_all_reduce,
tensor_model_parallel_all_reduce_custom,
)
from fastdeploy.model_executor.layers.utils import get_tensor
from fastdeploy.model_executor.utils import h2d_copy, slice_fn
from fastdeploy.platforms import current_platform
Expand Down Expand Up @@ -643,7 +646,10 @@ def forward(self, x: paddle.Tensor, gate: nn.Layer):
out = self.forward_normal(x, gate)

if self.reduce_results and self.tp_size > 1:
out = tensor_model_parallel_all_reduce(out, self.tp_group)
if current_platform.is_intel_hpu():
tensor_model_parallel_all_reduce_custom(out)
Comment thread
fmiao2372 marked this conversation as resolved.
else:
out = tensor_model_parallel_all_reduce(out, self.tp_group)
return out

def forward_chunked_moe(self, x: paddle.Tensor, gate: nn.Layer):
Expand Down
9 changes: 6 additions & 3 deletions fastdeploy/worker/hpu_model_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1120,10 +1120,11 @@ def warm_up_bucket(self) -> None:

max_prefill_length = self.cache_config.block_size + warmup_max_model_len
prefill_context_block_step = int(os.environ.get("CONTEXT_BLOCK_STEP_PREFILL", 1))
prefill_batchs.reverse()
prefill_length_with_contexts = list(range(self.cache_config.block_size, max_prefill_length, prefill_seq_step))
prefill_length_with_contexts.reverse()
Comment thread
fmiao2372 marked this conversation as resolved.
for prefill_batch in prefill_batchs:
for prefill_length_with_context in range(
self.cache_config.block_size, max_prefill_length, prefill_seq_step
):
for prefill_length_with_context in prefill_length_with_contexts:
if prefill_length_with_context * prefill_batch > self.scheduler_config.max_num_batched_tokens:
continue
for context_len in range(
Expand Down Expand Up @@ -1171,6 +1172,8 @@ def warm_up_bucket(self) -> None:
current_decode_block_num += decode_block_num_step

logger.info(f"warmup decode_batchs: {decode_batchs}, decode_block_nums: {decode_block_nums} start")
decode_batchs.reverse()
decode_block_nums.reverse()
for decode_batch in decode_batchs:
for decode_block_num in decode_block_nums:
if decode_block_num < decode_batch:
Expand Down
Loading