Repository navigation
Expert-parallel implementations for ep_dispatch_experts: all-reduce at inference, all-to-all in training - #49429
Conversation
|
The docs for this PR live here. All of your documentation changes will be reflected on that endpoint. The docs are available until 30 days after the last update. Pages changed in this PR: |
…ers into unify-resolve-plan
This reverts commit 3f1e4ab.
ep_dispatch_experts: all-reduce at inference, all-to-all in training
| ep_group = self.ep_mesh.get_group() | ||
| # every rank backpropagates through its own experts only: sum the replicated inputs' gradients | ||
| hidden_states = _AllReduceBackward.apply(hidden_states, ep_group) | ||
| if torch.is_grad_enabled() and top_k_weights.requires_grad: |
There was a problem hiding this comment.
how does that work with gradient checkpointing ? (since it's alternating no_grad and grad for the recompute)
There was a problem hiding this comment.
yes interesting case, with our default (use_reentrant=False) the first forward runs with grad enabled, so auto picks all-to-all both times. with use_reentrant=True the first forward runs under no_grad and gets the all-reduce, then the recompute gets the all-to-all: same function and same output shapes, so the gradients are right, only bf16 rounding differs between the two passes. i will add a test just in case
There was a problem hiding this comment.
why is this workaround needed in the first place? It seems a tad weird to me ngl
There was a problem hiding this comment.
not sur what you mean by workaround here
There was a problem hiding this comment.
Why do we need to check for if torch.is_grad_enabled() and top_k_weights.requires_grad:?
There was a problem hiding this comment.
bcz otherwise it's unnecessary to do the bwd, #49975 seems related
There was a problem hiding this comment.
yeah we can remove it now as the hook will only be attached during training
There was a problem hiding this comment.
not exactly, #49975 is about the param grad hooks of ReplicatedWithGradAllReduce. this one is the autograd fn on activations (router scores, ep inputs, mla rope..) and it still runs on every forward, so the check is what makes it free at inference (no autograd fn per layer during decode). so i would keep it
There was a problem hiding this comment.
can we at least make a proper helper on when we do need this?
There was a problem hiding this comment.
added _all_reduce_backward in parallel_utils, it skips the autograd fn when no grad can flow. all the tp/ep call sites use it now
|
@dacorvo you can now just pass |
| dist.all_reduce(grad, group=ctx.process_group) | ||
| return grad, None | ||
|
|
||
| def _all_reduce_backward(x, process_group): |
There was a problem hiding this comment.
We now have a function and class for _all_reduce_backward. Better call if torch.is_grad_enabled() and x.requires_grad: return _AllReduceBackward.apply(x, process_group) in ep implementation ?
There was a problem hiding this comment.
Done: the function is gone, and the 7 call sites (router scores and the all-reduce EP inputs in expert_parallel.py, plus the MoE TP experts, the router weights and MLA rope in tensor_parallel.py) check torch.is_grad_enabled() and x.requires_grad before _AllReduceBackward.apply. Its docstring says why.
There was a problem hiding this comment.
I like this more tbh but maybe we can hide it directly under _AllReduceBackward as a compromise, wdyt @3outeille
| f"{type(self).__name__} needs every rank of the expert group to hold the same batch (`ep_size == tp_size`)." | ||
| ) | ||
|
|
||
| def dispatch(self, hidden_states, top_k_index, top_k_weights): |
There was a problem hiding this comment.
My only concern is that AllReduceEP doesn't follow the dispatch + combine pattern which can bring confusion. Maybe we can just turn AllToAllEP and AllReduceEP class into simple functions to avoid having to deal with an explicit dispatch + combine pattern API ?
There was a problem hiding this comment.
AllReduceEP did follow it: its dispatch picked the routes this rank computes, and its combine summed the outputs. AutoEP was the one that didn't, since it overrode __call__ around both. To make that uniform, every implementation is now a callable, __call__(hidden_states, top_k_index, top_k_weights), with no dispatch / combine contract. They stay classes, since one can keep state across forwards (DeepEP's buffers, for example). A shared run_experts runs the local experts with their local params and keeps empty rows connected to backward. The docs' "Adding an implementation" example is a single __call__.
`AllReduceEP` had nothing to dispatch and `AutoEP` bypassed both steps, so the two-step contract didn't fit. An `EPImplementation` now implements `forward` and runs the local experts with a shared `run_experts` (local DTensor params, empty rows kept connected to backward). `AllToAllEP` keeps its two exchanges as private `_send` / `_receive` steps. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Every `EPImplementation` defines `__call__` (AutoEP already did), and the base class keeps `run_experts`, now owning the local DTensor params context. `_all_reduce_backward` is gone: its call sites check `torch.is_grad_enabled() and x.requires_grad` before `_AllReduceBackward.apply`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3outeille
left a comment
There was a problem hiding this comment.
nice work almost there, small changes and it should be good to go
- `EpDispatchExpertsParallel.ep_implementations` maps the three built-in names to their classes, replacing `EPImplementationInterface` / `ALL_EP_IMPLEMENTATIONS`; docs and the config docstring point to it. - `AllToAllEP`'s halves are `_dispatch_tokens` and `_combine_tokens` again. - `EpDispatchExpertsParallel.should_use_local_tensors` is back. - Tests, from review: one parametrized check per implementation and mode (train, inference, both checkpoint modes) against one process, a separate check of AutoEP's choice, and install_forward's refusals. Co-authored-by: Ferdinand Mom <47445085+3outeille@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7c1410e to
d876395
Compare
`_same_ranks` had one caller; the comparison of the TP and EP groups' ranks now sits where `shares_batch` is set. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
vasqu
left a comment
There was a problem hiding this comment.
Ok overall very nice state, chiming in to give my 2 cents but I think it's close to merge
| # before the mask (each slot has exactly one owning rank, so the sum is exact). | ||
| process_group = mesh.get_group() if mesh.ndim == 1 else mesh.get_group("tp") | ||
| if torch.is_grad_enabled() and router_scores.requires_grad: | ||
| process_group = mesh.get_group() if mesh.ndim == 1 else mesh.get_group("tp") |
There was a problem hiding this comment.
why did we change the order here? doesnt have any positive effect no?
There was a problem hiding this comment.
i made this "if grad: all reduce backward" a helper function before but we inlined it after, this is just the result of that revert, now it just aligns with the pattern without the process_group line in the middle:
if torch.is_grad_enabled() and x.requires_grad:
x = _AllReduceBackward.apply(x, process_group)There was a problem hiding this comment.
Is it expensive to check the process group? Ig its negligible, just wanna be sure it doesnt cause an unnecessary overhead on inference
| """ | ||
|
|
||
| # Set by implementations that only run when every rank of the EP group holds the same batch, e.g. `AllReduceEP`. | ||
| needs_shared_batch = False |
There was a problem hiding this comment.
ig deepgeemm and deep ep will also need it?
There was a problem hiding this comment.
no, only implementations that skip the token exchange need it. deep ep dispatches tokens like all_to_all so it works whatever batch each rank holds, and megamoe is its own style (megamoe_experts), not an EPImplementation
| def __init__(self, experts, ep_mesh, tp_mesh=None): | ||
| self.experts = experts | ||
| self.experts_forward = experts.forward # the experts' own forward, on local expert ids | ||
| self.num_local_experts = experts.num_experts |
There was a problem hiding this comment.
are we sure num experts is safe for all models? do we need something like in fp8 checking for attrs?
There was a problem hiding this comment.
yes, the experts behind @use_experts_implementation all have num_experts (the sentinel masking reads it), and grouped_gemm sharding already rewrites it to the local count
| needs_shared_batch = False | ||
|
|
||
| def __init__(self, experts, ep_mesh, tp_mesh=None): | ||
| self.experts = experts |
There was a problem hiding this comment.
why are we saving the experts?
There was a problem hiding this comment.
run_experts needs the module to swap its dtensor params for local ones (_use_local_dtensor_params(self.experts)), and an implementation with its own kernel would read the weights from it
| """ | ||
| Dispatch disjoint TP token slices to the experts' owners, then replicate the combined output on TP. | ||
| The ranks of a TP group keep disjoint slices of the batch they share, send each (token, expert) pair to the | ||
| expert's rank with an all-to-all and get the results back with a second one, then gather the slices back over TP. |
There was a problem hiding this comment.
so we always have a TP mesh no matter what, just to be sure I follow
There was a problem hiding this comment.
yeah it's a proper EP and it needs the tp_size to avoid computing tp_size time the same token. Cf the explanation here: #48865
There was a problem hiding this comment.
yep always there, size 1 without tp. made tp_mesh required and dropped the None branches
| def ep_forward(hidden_states, top_k_index, top_k_weights): | ||
| if isinstance(hidden_states, DTensor): | ||
| hidden_states = hidden_states.to_local() | ||
| if isinstance(top_k_weights, DTensor): | ||
| top_k_weights = top_k_weights.to_local() | ||
| return implementation(hidden_states, top_k_index, top_k_weights).to(hidden_states.dtype) |
There was a problem hiding this comment.
feels a bit weird, maybe we can reuse the local_params style within tp?
There was a problem hiding this comment.
these are the inputs not the params, params already go through _use_local_dtensor_params in run_experts (same as local_params). the input unwrap is carried over from #48865, MoeExpertsParallel does the same in its pre-forward. can move both to a shared helper in a follow up
There was a problem hiding this comment.
yea that would be nice, indeed it's the input mb
There was a problem hiding this comment.
any call to parallel function should in general / priority be done here
There was a problem hiding this comment.
done, the dtensor unwrapping and the local params both live in ep_forward now, the implementations only move tokens and call the experts
| # Dispatch hooks need both meshes to redistribute tokens between TP and EP ranks. | ||
| if style_name == "ep_dispatch_experts": | ||
| style.install_forward(module, ep_mesh=ep_mesh, tp_mesh=tp_mesh) | ||
| style.install_forward(module, ep_mesh, tp_mesh=tp_mesh, ep_implementation=ep_implementation) |
There was a problem hiding this comment.
lets keep kwargs styöe on ep mesh?
There was a problem hiding this comment.
yea looks like we should do a proper helper
There was a problem hiding this comment.
done, _all_reduce_backward
| assert (all_reduce_call.called, all_to_all_call.called) == (all_reduce, not all_reduce), (tp_mesh, grad) | ||
|
|
||
|
|
||
| class _ScaledExperts(torch.nn.Module): |
There was a problem hiding this comment.
why an extra one and not include it as optional kwarg within dummy experts?
There was a problem hiding this comment.
merged them, _DummyExperts(scales=None)
| output = checkpoint(forward, experts, *inputs, use_reentrant=False) | ||
| elif mode == "reentrant_checkpoint": | ||
| output = checkpoint(forward, experts, *inputs, use_reentrant=True) |
There was a problem hiding this comment.
imo reentrant true behaves very weird and is actively discouraged for the most part; do we really want to test it?
There was a problem hiding this comment.
it is exactly the gradient checkpointing case from above: auto runs all_reduce on the no_grad first pass and all_to_all on the recompute, so it is the one place both implementations get mixed. users can still pass use_reentrant=True so i would keep it, it is cheap
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…xperts class in tests Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
vasqu
left a comment
There was a problem hiding this comment.
Ok I think this is ready, some last comments but nothing major anymore
| | :--- | :--- | :--- | | ||
| | `"all_to_all"`, see [token dispatch](#token-dispatch) | Works for any layout, including `ep_size > tp_size`. Only the routed tokens move. | Two all-to-all collectives and one host sync per layer, so it does not capture in a CUDA graph. | | ||
| | `"all_reduce"` | No tokens move and no host sync, so decode captures in a CUDA graph. | Requires `ep_size=tp_size`. Every rank runs its experts over the whole batch, so its cost grows with the batch. | | ||
| | `"auto"`, the default | Runs `"all_reduce"` at inference when the ranks of an expert group share one batch, and `"all_to_all"` otherwise. | | |
There was a problem hiding this comment.
Disadvantage might be that this is not always ideal and your workload can benefit from custom implementations at times
| | `"all_reduce"` | No tokens move and no host sync, so decode captures in a CUDA graph. | Requires `ep_size=tp_size`. Every rank runs its experts over the whole batch, so its cost grows with the batch. | | ||
| | `"auto"`, the default | Runs `"all_reduce"` at inference when the ranks of an expert group share one batch, and `"all_to_all"` otherwise. | | | ||
|
|
||
| `ep_implementation` only chooses within the parallel sizes set in [`DistributedConfig`]. When the ranks of an expert group hold different batches, for example `tp_size=1` with `fsdp_size=ep_size`, `"all_to_all"` is the only implementation that applies. |
| expert_output = self.run_experts(tokens, expert_ids, weights) | ||
| return ... # the weighted expert output for each token of the batch this rank holds | ||
|
|
||
| EpDispatchExpertsParallel.ep_implementations["my_ep"] = MyEP |
There was a problem hiding this comment.
do we not have a proper function to register?
| distributed_config = DistributedConfig(tp_size=1, fsdp_size=8, ep_size=8, ep_implementation="my_ep") | ||
| ``` | ||
|
|
||
| Gradients must reach every rank that holds a token, routing weights included, so wrap a collective that autograd does not differentiate in a `torch.autograd.Function`. |
There was a problem hiding this comment.
Also more appropriate as a note imo
| Number of devices owning distinct expert shards. Defaults to 1. Set it explicitly to enable EP. Must be | ||
| a multiple of `tp_size` and divide `fsdp_size * tp_size`. All-reduce expert plans require | ||
| `ep_size=tp_size`; token dispatch (`"ep_dispatch_experts"`) also allows `ep_size > tp_size`. | ||
| `ep_size=tp_size`; `"ep_dispatch_experts"` also allows `ep_size > tp_size`, with all-to-all token dispatch. |
There was a problem hiding this comment.
Imo this whole last sentence needs a rewrite no? ep dispatch experts is non telling with this change as the underlying implementation is important and id rather we go over that no?
| dist.all_reduce(grad, group=ctx.process_group) | ||
| return grad, None | ||
|
|
||
| def _all_reduce_backward(x, process_group): |
There was a problem hiding this comment.
I like this more tbh but maybe we can hide it directly under _AllReduceBackward as a compromise, wdyt @3outeille
| return self.experts_forward(tokens, expert_ids, weights) | ||
|
|
||
|
|
||
| class _AllToAllState(NamedTuple): |
There was a problem hiding this comment.
prolly unnecessary worry but is it safe for older torch versions under compile?
| """Each (token, expert) pair to the expert's rank: the rows this rank's experts run on, their local expert ids | ||
| and weights, and what `_combine_tokens` needs to bring the results back.""" | ||
| num_tokens = hidden_states.size(0) | ||
| if self.tp_mesh.size() > 1: |
There was a problem hiding this comment.
not this PR necessarily but I do feel like we can make the input handling under here more pretty but no worries for now
3outeille
left a comment
There was a problem hiding this comment.
Hiding _all_reduce_backward inside _AllReduceBackward
|
|
||
| Apply it through `_all_reduce_backward`, which skips it when no gradient can flow. | ||
| """ | ||
|
|
There was a problem hiding this comment.
| @classmethod | |
| def apply(cls, x, process_group): | |
| if torch.is_grad_enabled() and x.requires_grad: | |
| return super().apply(x, process_group) | |
| return x |
| dist.all_reduce(grad, group=ctx.process_group) | ||
| return grad, None | ||
|
|
||
| def _all_reduce_backward(x, process_group): |
There was a problem hiding this comment.
| def _all_reduce_backward(x, process_group): |
…face registry, doc notes Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
ArthurZucker
left a comment
There was a problem hiding this comment.
one abstraction not super sure it makes sense to register fully or just inherit for now but otherwise yes, ty
| An implementation subclasses `EPImplementation` and registers under the name `ep_implementation` selects. It is built once per experts module, so it can keep state such as communication buffers. Calling it with the batch this rank holds, with global expert ids, returns the weighted expert output for each of its tokens. It runs this rank's experts with `run_experts`, on the rows, local expert ids and weights it brought them. | ||
|
|
||
| ```py | ||
| from transformers.distributed.expert_parallel import EPImplementation, EPImplementationInterface |
There was a problem hiding this comment.
| from transformers.distributed.expert_parallel import EPImplementation, EPImplementationInterface | |
| from transformers.distributed.expert_parallel import ExpertImplementation, ExpertImplementationInterface |
let's use good naming
There was a problem hiding this comment.
then it has to be expert prallel implementation, because expert implementation is already a thing (grouped/batched/sonicmoe/etc..)
There was a problem hiding this comment.
on second thought kept EPImplementation: everything EP here is ep_ / EP (ep_size, ep_plan, ep_implementation, AllToAllEP, EpDispatchExpertsParallel), and it stays clear of experts_implementation
| expert_output = self.run_experts(tokens, expert_ids, weights) | ||
| return ... # the weighted expert output for each token of the batch this rank holds | ||
|
|
||
| EPImplementationInterface.register("my_ep", MyEP) |
There was a problem hiding this comment.
probably not needed if post_init form super registers with STR of name
There was a problem hiding this comment.
i'd rather keep it explicit: registering on subclass definition means just defining a class mutates a global registry (even in tests). EPImplementationInterface is a GeneralInterface, same as AttentionInterface / ExpertsInterface, so registering an ep implementation looks like registering an attention or experts one
| `ep_size > 1`, and its rules take precedence over `tp_plan` rules for the same modules. An | ||
| `"ep_dispatch_experts"` rule selects all-to-all token dispatch instead of router masking and all-reduce. | ||
| `"ep_dispatch_experts"` rule lets `ep_implementation` pick how tokens reach the experts. | ||
| ep_implementation (`str`, *optional*, defaults to `"auto"`): |
There was a problem hiding this comment.
expert_implementation
There was a problem hiding this comment.
kept ep_implementation, same reason as the class name: expert_implementation would be one letter off experts_implementation (grouped_mm, batched_mm, ...) which is a different thing, and ep_ matches ep_size / ep_plan
|
|
||
|
|
||
| class EpDispatchExpertsParallel(ExpertParallelLayer): | ||
| class EPImplementation: |
There was a problem hiding this comment.
| class EPImplementation: | |
| class ExpertImplementation: |
There was a problem hiding this comment.
kept EPImplementation, see the docs thread: ExpertImplementation would read as an experts_implementation (grouped_mm etc.), and the rest of the EP code uses the EP prefix
| # Set by implementations that only run when every rank of the EP group holds the same batch, e.g. `AllReduceEP`. | ||
| needs_shared_batch = False | ||
|
|
||
| def __init__(self, experts, ep_mesh, tp_mesh): |
There was a problem hiding this comment.
type hint on the expert at least !
| _global_mapping = {"auto": AutoEP, "all_to_all": AllToAllEP, "all_reduce": AllReduceEP} | ||
|
|
||
|
|
||
| class EpDispatchExpertsParallel(ExpertParallelLayer): |
There was a problem hiding this comment.
not super sure we need AutoEP, EPImplementationInterfac.
There's not much shared between both, and some mixing of parallel stuff
There was a problem hiding this comment.
kept both: auto is the default path, as its own small class it's readable and tested on its own, instead of branching inside install_forward. the interface is for the same reason as the other thread (same pattern as ExpertsInterface). what did move is the local params context, see below
|
|
||
| def run_experts(self, tokens, expert_ids, weights): | ||
| """This rank's experts on `tokens`, routed to `expert_ids` (local ids) with `weights`.""" | ||
| with _use_local_dtensor_params(self.experts): |
There was a problem hiding this comment.
well, it's sure is weird to define this whole interface just to call _use_local_dtensor_params.
I mean, this should be done by the ExpertParallel.forward ?
There was a problem hiding this comment.
yep moved it, ep_forward now runs the implementation inside the style's context_around_forward (local params), and run_experts is gone, implementations call experts_forward directly
| """This rank's experts on `tokens`, routed to `expert_ids` (local ids) with `weights`.""" | ||
| with _use_local_dtensor_params(self.experts): | ||
| if tokens.shape[0] == 0: | ||
| # Keep the empty rows connected to backward, so the reverse exchange and FSDP's reduction still run. |
There was a problem hiding this comment.
correct me if I am wrong, only a single class uses it as off today
There was a problem hiding this comment.
right, only all-to-all can receive zero rows, moved it into AllToAllEP
| def ep_forward(hidden_states, top_k_index, top_k_weights): | ||
| if isinstance(hidden_states, DTensor): | ||
| hidden_states = hidden_states.to_local() | ||
| if isinstance(top_k_weights, DTensor): | ||
| top_k_weights = top_k_weights.to_local() | ||
| return implementation(hidden_states, top_k_index, top_k_weights).to(hidden_states.dtype) |
There was a problem hiding this comment.
any call to parallel function should in general / priority be done here
…y-rows case in `AllToAllEP` Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
"ep_dispatch_experts"reaches its experts through an implementation thatDistributedConfig.ep_implementationpicks. Inference withep_size == tp_sizeno longer pays an all-to-all and a host sync per layer, so decode captures in a CUDA graph.Implementations
ep_implementation"all_to_all""all_reduce"ep_size == tp_size"auto", the default"all_reduce"at inference on a shared batch,"all_to_all"otherwise.The implementations live in
ALL_EP_IMPLEMENTATIONS. A new one subclassesEPImplementation, implementsdispatchandcombine, and registers under a name.Changes
distributed/expert_parallel.pyEPImplementation,AllToAllEP(the former dispatch code),AllReduceEP,AutoEPandALL_EP_IMPLEMENTATIONS.EpDispatchExpertsParalleldelegates to them and rejects an unknown name.distributed/configuration_utils.pyep_implementationfield,"auto"by default.distributed/mixin.pyep_implementationtoapply_expert_parallelism.distributed/parallel_utils.py_same_ranks, and_all_reduce_backward. It runs the backward-only all-reduce only when a gradient can flow, at every TP and EP site that used it: MoE TP experts, MLA rope, router scores and EP inputs. That saves ~1.8 µs of host time per call in eager decode.docs/source/en/expert_parallelism.mdtests/expert_parallel/test_expert_parallel.py"auto"choice, gradient checkpointing in both modes, unknown and custom names, and the shared-batch check.Validation
tests/expert_parallel/test_expert_parallel.py, 2 ranks on gloo🤖 Generated with Claude Code