-
Notifications
You must be signed in to change notification settings - Fork 7.4k
[core] Shard tensor-parallel checkpoints on load and save #14544
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.
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
base: main
Are you sure you want to change the base?
Changes from all commits
acdf4bf
0b9686b
40ddb53
0760934
842c643
949cbdc
eb3f7f3
a754bbe
6055292
fd31c34
9511143
53fd823
4112d18
f741782
1b4d1d7
4e9a927
0cca9a3
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -436,43 +436,42 @@ pipeline = DiffusionPipeline.from_pretrained( | |
|
|
||
| [Tensor parallelism](https://huggingface.co/spaces/nanotron/ultrascale-playbook?section=tensor_parallelism) shards the weight matrices of a model across devices. Each device holds a column-wise (`"colwise"`) or row-wise (`"rowwise"`) slice of each layer, computes a partial result, and an `AllReduce`/`AllGather` at the layer boundary reconstructs the full output. Unlike context parallelism, it reduces the per-device *weight* memory, which is useful for models that do not fit on a single device. | ||
|
|
||
| Pass a [`TensorParallelConfig`] to [`~ModelMixin.enable_parallelism`]. `tp_degree` is the number of devices to shard across and must divide the model's number of attention heads. The model must define a `_tp_plan` (a flat mapping of module-name globs to a `"colwise"`/`"rowwise"` style). | ||
| Pass a [`TensorParallelConfig`] to the `parallel_config` argument of the model's [`~ModelMixin.from_pretrained`]. `tp_degree` is the number of devices to shard across and must divide the model's number of attention heads. The model must define a `_tp_plan` (a flat mapping of module-name globs to a `"colwise"`/`"rowwise"` style). | ||
|
|
||
| Loading this way shards the checkpoint *while reading it*: each rank reads only its own slice of each sharded weight and places it straight onto its own device. Nothing full-size is ever materialized, so per-rank memory falls as `tp_degree` rises. | ||
|
|
||
| ```py | ||
| import torch | ||
| from torch import distributed as dist | ||
| from diffusers import DiffusionPipeline, TensorParallelConfig | ||
| from diffusers import DiffusionPipeline, Flux2Transformer2DModel, TensorParallelConfig | ||
|
|
||
| def setup_distributed(): | ||
| if not dist.is_initialized(): | ||
| dist.init_process_group(backend="nccl") | ||
| rank = dist.get_rank() | ||
| def main(): | ||
| dist.init_process_group(backend="nccl") | ||
| rank, world_size = dist.get_rank(), dist.get_world_size() | ||
| device = torch.device(f"cuda:{rank}") | ||
| torch.cuda.set_device(device) | ||
| return device | ||
|
|
||
| def main(): | ||
| device = setup_distributed() | ||
| world_size = dist.get_world_size() | ||
| # Each rank reads only its own shard of every planned weight, straight onto `cuda:rank`. | ||
| transformer = Flux2Transformer2DModel.from_pretrained( | ||
| "black-forest-labs/FLUX.2-dev", | ||
| subfolder="transformer", | ||
| torch_dtype=torch.bfloat16, | ||
| parallel_config=TensorParallelConfig(tp_degree=world_size), | ||
| ) | ||
|
|
||
| pipeline = DiffusionPipeline.from_pretrained( | ||
| "black-forest-labs/FLUX.2-dev", torch_dtype=torch.bfloat16 | ||
| ) # weights stay on CPU | ||
|
|
||
| # Shard the transformer first, then move only each rank's slice onto the accelerator. | ||
| pipeline.transformer.enable_parallelism(config=TensorParallelConfig(tp_degree=world_size)) | ||
| pipeline.transformer.to(device) | ||
|
|
||
| # Move the remaining, non-sharded components onto the accelerator individually. | ||
| "black-forest-labs/FLUX.2-dev", transformer=transformer, torch_dtype=torch.bfloat16 | ||
| ) | ||
| # The transformer is already on its device; move the remaining components individually. Do not call | ||
| # `pipeline.to(device)` — that would move every rank's shards onto the same device. | ||
| pipeline.text_encoder.to(device) | ||
| pipeline.vae.to(device) | ||
|
|
||
| generator = torch.Generator().manual_seed(42) | ||
| image = pipeline(prompt="a cat holding a sign that says hello", generator=generator).images[0] | ||
| if dist.get_rank() == 0: | ||
| if rank == 0: | ||
| image.save("output.png") | ||
| if dist.is_initialized(): | ||
| dist.destroy_process_group() | ||
| dist.destroy_process_group() | ||
|
|
||
| if __name__ == "__main__": | ||
| main() | ||
|
|
@@ -484,6 +483,10 @@ torchrun --nproc-per-node 4 tensor_parallel_flux.py | |
|
|
||
| `tp_degree` is taken from `world_size` above, so `--nproc-per-node 4` shards the transformer across 4 devices. | ||
|
|
||
| A tensor-parallel `parallel_config` cannot be combined with `device_map`, `quantization_config`, `low_cpu_mem_usage=False`, `use_flashpack=True`, or non-safetensors weights; each raises rather than quietly falling back to loading the full checkpoint. Tensor parallelism also cannot be combined with quantization, offloading, or LoRA adapters at all — the parameters it shards have to be plain parameters owned by the model — so those raise however the model is sharded. To shard a model that is already in memory, call [`~ModelMixin.enable_parallelism`] with the same config instead — that loads everything first and reshards it, so it costs full checkpoint memory on every rank. | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Interesting that we cannot load TP with quantization. Do we know why?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It's mostly about scoping... I didn't want to handle "quantization + TP" sharding in the scale of this PR. IMO, it could be possible if we shard and then quantize, and then according to the quantization configs and the backend, some might work, some doesn't... I would rather raise for now, we could consider supporting the combo properly if the community shows interest. |
||
|
|
||
| Saving a tensor-parallel model isn't supported yet, and [`~ModelMixin.save_pretrained`] raises on one. Save the model before sharding it. | ||
|
|
||
| ### Writing a tensor parallelism plan | ||
|
|
||
| Tensor parallelism only works on models that define a `_tp_plan`, a flat class attribute mapping module-name globs to a sharding style. Writing one is mostly a matter of pairing each projection that *expands* the hidden dimension with the projection that *contracts* it back. | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oh really! This is very cool. Could we also present a small comparison between the loading time with and without this way of loading?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Here's a quick benchmark:
Synthetic checkpoint (FLUX.1-shaped, 1.33B params, 2.65GB bf16):
from_pretrained(parallel_config=...))from_pretrained+enable_parallelism)from_pretrained(parallel_config=...))from_pretrained+enable_parallelism)