Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
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
45 changes: 24 additions & 21 deletions docs/source/en/training/distributed_inference.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Copy link
Copy Markdown
Member

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?

Copy link
Copy Markdown
Contributor Author

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):

mode tp_degree load time peak GPU/rank peak CPU/rank
streamed (from_pretrained(parallel_config=...)) 2 1.92s ± 0.11 1.71GB 2.70GB
materialize+reshard (from_pretrained + enable_parallelism) 2 3.06s ± 0.02 1.69GB 4.08GB
streamed (from_pretrained(parallel_config=...)) 4 1.45s ± 0.03 1.23GB 2.20GB
materialize+reshard (from_pretrained + enable_parallelism) 4 3.29s ± 0.02 1.23GB 4.08GB
  • Speed: streaming is ~37% faster at tp_degree=2, ~56% faster at tp_degree=4 on the synthetic checkpoint (1.92s vs 3.06s, 1.45s vs 3.29s).
  • GPU memory: basically the same between the two.
  • CPU/host memory: 34% less at tp_degree=2, 46% less at tp_degree=4 (2.70GB vs 4.08GB, 2.20GB vs 4.08GB).


```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()
Expand All @@ -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.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interesting that we cannot load TP with quantization. Do we know why?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The 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.
Expand Down
Loading
Loading