Skip to content

Commit 64cbf19

Browse files
committed
Allow FSDP strategy on CPU accelerator
This enables using FSDP strategy with `accelerator="cpu"`, which is useful for CPU-based FSDP checkpoint benchmarking and testing. Also updated CHANGELOG.md and added tests. TAG=agy CONV=0b443e3e-6fbb-468a-9f11-b99f211cf34b
1 parent fbdf042 commit 64cbf19

4 files changed

Lines changed: 134 additions & 13 deletions

File tree

src/lightning/pytorch/CHANGELOG.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
2222

2323
- Added support for remote storage (fsspec URLs) when saving and loading distributed checkpoints with `ModelParallelStrategy` ([#21797](https://github.com/Lightning-AI/pytorch-lightning/issues/21797))
2424

25+
- Added support for using the `FSDPStrategy` on CPU (`accelerator="cpu"`) ([#21812](https://github.com/Lightning-AI/pytorch-lightning/issues/21812))
26+
2527
### Changed
2628

2729
-

src/lightning/pytorch/trainer/connectors/accelerator_connector.py

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
from lightning.fabric.utilities.imports import _IS_INTERACTIVE
3434
from lightning.pytorch.accelerators import AcceleratorRegistry
3535
from lightning.pytorch.accelerators.accelerator import Accelerator
36+
from lightning.pytorch.accelerators.cpu import CPUAccelerator
3637
from lightning.pytorch.accelerators.cuda import CUDAAccelerator
3738
from lightning.pytorch.accelerators.mps import MPSAccelerator
3839
from lightning.pytorch.accelerators.xla import XLAAccelerator
@@ -430,11 +431,14 @@ def _check_strategy_and_fallback(self) -> None:
430431

431432
if (
432433
strategy_flag in FSDPStrategy.get_registered_strategies() or type(self._strategy_flag) is FSDPStrategy
433-
) and not (self._accelerator_flag in ("cuda", "gpu") or isinstance(self._accelerator_flag, CUDAAccelerator)):
434+
) and not (
435+
self._accelerator_flag in ("cuda", "gpu", "cpu")
436+
or isinstance(self._accelerator_flag, (CUDAAccelerator, CPUAccelerator))
437+
):
434438
raise ValueError(
435-
f"The strategy `{FSDPStrategy.strategy_name}` requires a GPU accelerator, but received "
439+
f"The strategy `{FSDPStrategy.strategy_name}` requires a GPU or CPU accelerator, but received "
436440
f"`accelerator={self._accelerator_flag!r}`. Please set `accelerator='cuda'`, `accelerator='gpu'`,"
437-
" or pass a `CUDAAccelerator()` instance to use FSDP."
441+
" `accelerator='cpu'`, or pass a `CUDAAccelerator()`/`CPUAccelerator()` instance to use FSDP."
438442
)
439443
if strategy_flag in _DDP_FORK_ALIASES and "fork" not in torch.multiprocessing.get_all_start_methods():
440444
raise ValueError(

tests/tests_pytorch/strategies/test_fsdp.py

Lines changed: 121 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,15 +14,17 @@
1414
import torch.nn as nn
1515
from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload, FullyShardedDataParallel, MixedPrecision
1616
from torch.distributed.fsdp.wrap import ModuleWrapPolicy, always_wrap_policy, size_based_auto_wrap_policy, wrap
17+
from torch.utils.data import DataLoader
1718
from torchmetrics import Accuracy
1819

1920
from lightning.fabric.plugins.environments import LightningEnvironment
2021
from lightning.fabric.strategies.fsdp import _is_sharded_checkpoint
2122
from lightning.fabric.utilities.imports import _TORCH_GREATER_EQUAL_2_2, _TORCH_GREATER_EQUAL_2_3
2223
from lightning.fabric.utilities.load import _load_distributed_checkpoint
23-
from lightning.pytorch import Trainer
24+
from lightning.pytorch import Trainer, seed_everything
25+
from lightning.pytorch.accelerators.cpu import CPUAccelerator
2426
from lightning.pytorch.callbacks import ModelCheckpoint
25-
from lightning.pytorch.demos.boring_classes import BoringModel
27+
from lightning.pytorch.demos.boring_classes import BoringModel, RandomDataset
2628
from lightning.pytorch.plugins import HalfPrecision
2729
from lightning.pytorch.plugins.precision.fsdp import FSDPPrecision
2830
from lightning.pytorch.strategies import FSDPStrategy
@@ -194,12 +196,123 @@ def _assert_save_equality(trainer, ckpt_path, cls=TestFSDPModel):
194196
assert torch.equal(ddp_param, shard_param)
195197

196198

197-
def test_invalid_on_cpu(tmp_path, cuda_count_0):
198-
"""Test to ensure that we raise Misconfiguration for FSDP on CPU."""
199-
with pytest.raises(ValueError, match="The strategy `fsdp` requires a GPU accelerator"):
200-
trainer = Trainer(accelerator="cpu", default_root_dir=tmp_path, fast_dev_run=True, strategy="fsdp")
201-
assert isinstance(trainer.strategy, FSDPStrategy)
202-
trainer.strategy.setup_environment()
199+
def test_fsdp_on_cpu_allowed(tmp_path, cuda_count_0):
200+
"""FSDP on CPU is allowed (enables CPU-based FSDP checkpointing)."""
201+
trainer = Trainer(accelerator="cpu", default_root_dir=tmp_path, fast_dev_run=True, strategy="fsdp")
202+
assert isinstance(trainer.strategy, FSDPStrategy)
203+
assert isinstance(trainer.accelerator, CPUAccelerator)
204+
205+
206+
class _CPUFSDPTrainableModel(BoringModel):
207+
"""Overfits a fixed dataset so that the training loss must drop if FSDP actually trains the model.
208+
209+
If ``params_to_compare`` is given, the (already resharded) parameters are asserted to match it at the start of
210+
training. This is used to verify that a checkpoint written after training reads back with identical weights.
211+
"""
212+
213+
def __init__(self, params_to_compare: Optional[list[torch.Tensor]] = None):
214+
super().__init__()
215+
self.layer: Optional[nn.Module] = None
216+
self.losses: list[float] = []
217+
self.params_to_compare = params_to_compare
218+
219+
def configure_model(self) -> None:
220+
if self.layer is None:
221+
self.layer = torch.nn.Sequential(torch.nn.Linear(32, 4), torch.nn.ReLU(), torch.nn.Linear(4, 2))
222+
if isinstance(self.layer, FullyShardedDataParallel):
223+
return
224+
# wrap the two Linear layers and the container so their parameters are really flattened/sharded by FSDP
225+
self.layer[0] = wrap(self.layer[0])
226+
self.layer[2] = wrap(self.layer[2])
227+
self.layer = wrap(self.layer)
228+
229+
def configure_optimizers(self):
230+
return torch.optim.AdamW(self.layer.parameters(), lr=0.15)
231+
232+
def training_step(self, batch, batch_idx):
233+
loss = self.step(batch)
234+
self.losses.append(loss.detach().item())
235+
return {"loss": loss}
236+
237+
def on_train_batch_end(self, outputs, batch, batch_idx):
238+
# FSDP must be genuinely engaged here, not a silent fallback to plain CPU training
239+
assert isinstance(self.layer, FullyShardedDataParallel)
240+
241+
def on_train_start(self):
242+
# when resuming, the checkpoint has been loaded and resharded by now: the local shards must match exactly
243+
if self.params_to_compare is None:
244+
return
245+
for expected, actual in zip(self.params_to_compare, self.trainer.model.parameters()):
246+
torch.testing.assert_close(expected, actual, atol=0, rtol=0, equal_nan=True)
247+
248+
def train_dataloader(self):
249+
# a small, fixed dataset the model can overfit, so the loss drop is deterministic and not flaky
250+
return DataLoader(RandomDataset(32, 16), batch_size=8)
251+
252+
def val_dataloader(self):
253+
return DataLoader(RandomDataset(32, 16), batch_size=8)
254+
255+
256+
@pytest.mark.filterwarnings("ignore::FutureWarning")
257+
@RunIf(standalone=True, skip_windows=True)
258+
@pytest.mark.parametrize("state_dict_type", ["full", "sharded"])
259+
def test_fsdp_cpu_trainable(state_dict_type, tmp_path):
260+
"""FSDP on CPU across 2 ranks trains and round-trips a checkpoint.
261+
262+
Exercises both the ``full`` (single-file) and ``sharded`` (per-rank directory) checkpoint formats: the module is
263+
genuinely FSDP-wrapped, the loss drops sharply over 15 epochs, and the checkpoint written after training reads back
264+
with identical sharded parameters.
265+
"""
266+
267+
def _make_trainer(max_epochs: int) -> Trainer:
268+
return Trainer(
269+
accelerator="cpu",
270+
devices=2,
271+
strategy=FSDPStrategy(state_dict_type=state_dict_type),
272+
max_epochs=max_epochs,
273+
default_root_dir=tmp_path,
274+
enable_checkpointing=False,
275+
logger=False,
276+
enable_progress_bar=False,
277+
enable_model_summary=False,
278+
)
279+
280+
seed_everything(42)
281+
model = _CPUFSDPTrainableModel()
282+
trainer = _make_trainer(max_epochs=15)
283+
trainer.fit(model)
284+
285+
# the model was really trained under a 2-rank sharding regime with FSDP engaged (also asserted live per step)
286+
assert trainer.strategy.world_size == 2
287+
assert isinstance(model.layer, FullyShardedDataParallel)
288+
289+
# trainable: the loss dropped by more than 10x between the start and the end of training
290+
assert len(model.losses) > 5
291+
first = sum(model.losses[:3]) / 3
292+
last = sum(model.losses[-3:]) / 3
293+
assert last < 0.1 * first, f"loss did not drop enough: first={first:.4f} last={last:.4f}"
294+
295+
# --- checkpoint write ---
296+
# broadcast one path so every rank writes to (and later reads from) the same location
297+
ckpt_path = Path(trainer.strategy.broadcast(str(tmp_path / "checkpoint")))
298+
trainer.save_checkpoint(ckpt_path)
299+
300+
# the two formats must produce genuinely different on-disk layouts
301+
if state_dict_type == "sharded":
302+
assert ckpt_path.is_dir() and _is_sharded_checkpoint(ckpt_path)
303+
else:
304+
assert ckpt_path.is_file() and not _is_sharded_checkpoint(ckpt_path)
305+
306+
# snapshot the trained local shards on this rank for a read-back comparison
307+
trained_params = deepcopy(list(trainer.model.parameters()))
308+
309+
# --- checkpoint read ---
310+
# a fresh model + trainer must load the checkpoint and recover the exact trained parameters; the comparison runs in
311+
# `on_train_start`, once the checkpoint has been loaded and resharded (one extra epoch guarantees the hook fires)
312+
seed_everything(42)
313+
reloaded_model = _CPUFSDPTrainableModel(params_to_compare=trained_params)
314+
resumed_trainer = _make_trainer(max_epochs=15)
315+
resumed_trainer.fit(reloaded_model, ckpt_path=ckpt_path)
203316

204317

205318
def test_custom_mixed_precision():

tests/tests_pytorch/trainer/connectors/test_accelerator_connector.py

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -563,8 +563,10 @@ def test_strategy_choice_ddp_cpu_slurm(cuda_count_0, strategy):
563563

564564

565565
def test_check_fsdp_strategy_and_fallback():
566-
with pytest.raises(ValueError, match="The strategy `fsdp` requires a GPU accelerator"):
567-
Trainer(accelerator="cpu", strategy="fsdp")
566+
# FSDP on CPU is now allowed (enables CPU-based FSDP checkpoint benchmarking).
567+
trainer = Trainer(accelerator="cpu", strategy="fsdp")
568+
assert isinstance(trainer.strategy, FSDPStrategy)
569+
assert isinstance(trainer.accelerator, CPUAccelerator)
568570

569571
class FSDPStrategySubclass(FSDPStrategy):
570572
pass

0 commit comments

Comments
 (0)