VhOddlZddlmZmZddlmZddlmZmZddl Z ddl m Z ddl m cmcmcmZddlmZddgZGddeZGd deZy) N)ABCabstractmethod)Iterable)OptionalUnion)not_none ModelAveragerPeriodicModelAveragercHeZdZdZddeej fdZedZ y)r aHBase class for all model averagers. Args: process_group: The process group to be used for all-reduce. If ``None``, the default process group, which is created by :func:`torch.distributed.init_process_group`, will be used. (default: ``None``) N process_groupcj||n"ttjj|_d|_y)Nr) _not_nonedistgroupWORLDr step)selfr s f/home/dcms/DCMS/lib/python3.12/site-packages/torch/distributed/algorithms/model_averaging/averagers.py__init__zModelAverager.__init__s,*6MIdjjFVFV`_, by running :class:`~torch.nn.DistributedDataParallel` (DDP) using the subgroups created by :meth:`~torch.distributed.new_subgroups`. Args: period (int): The number of steps per model averaging. Usually the period should be greater than ``1`` to reduce the communication cost. Otherwise, only DDP needs to be used. warmup_steps (int): The number of warm-up steps. During this stage, model averaging is skipped. process_group: The process group to be used for all-reduce. If ``None``, the default process group, which is created by :func:`torch.distributed.init_process_group`, will be used. (default: ``None``) Example:: >>> # xdoctest: +SKIP("undefined variables") >>> import torch >>> import torch.distributed as dist >>> import torch.distributed.algorithms.ddp_comm_hooks.post_localSGD_hook as post_localSGD >>> import torch.distributed.algorithms.model_averaging.averagers as averagers >>> import torch.nn as nn >>> >>> dist.init_process_group("nccl", rank=rank, world_size=16) >>> torch.cuda.set_device(rank) >>> module = nn.Linear(1, 1, bias=False).cuda() >>> model = nn.parallel.DistributedDataParallel( >>> module, device_ids=[rank], output_device=rank >>> ) >>> # Register a post-localSGD communication hook. >>> state = PostLocalSGDState(process_group=None, subgroup=None, start_localSGD_iter=100) >>> model.register_comm_hook(state, post_localSGD_hook) >>> >>> # In the first 100 steps, run global gradient averaging like normal DDP at every step. >>> # After 100 steps, run model averaging every 4 steps. >>> # Note that ``warmup_steps`` must be the same as ``start_localSGD_iter`` used in ``PostLocalSGDState``. >>> averager = averagers.PeriodicModelAverager(period=4, warmup_steps=100) >>> for step in range(0, 200): >>> optimizer.zero_grad() >>> loss = loss_fn(output, labels) >>> loss.backward() >>> optimizer.step() >>> # Will average model parameters globally every 4 steps. Thus, >>> # inter-node communication only occurs every 4 iterations after >>> # the initial ``warmup_steps`` period. >>> averager.average_parameters(model.parameters()) r ct|||dkr td||_|dkr td|dk(rt j d||_y)Nrz3Arg ``warmup_steps`` must be a non-negative number.z(Arg ``period`` must be a positive value.aWhen period is 1, no need to use model averaging because the communication cost of all-reducing parameters will be no less than the cost of all-reducing gradients by DistributedDataParallel in the backward pass. Therefore, only DistributedDataParallel should be used for this case.)superr ValueError warmup_stepswarningswarnperiod)rr+r(r __class__s rrzPeriodicModelAverager.__init__Zsc ' ! RS S( A:GH H q[ MMH   rrc|j|jk\rR|j|jz |jzdk(r)tj|t |j |xjdz c_y)aQ Averages parameters or parameter groups of an optimizer if ``step`` is no less than ``warmup_steps``. Can be divided by ``period``, where ``step`` is increased by 1 at each iteration in the training loop. Args: params: The parameters of a model or parameter groups of an optimizer. rr%N)rr(r+utils&average_parameters_or_parameter_groupsrr rs rrz(PeriodicModelAverager.average_parameterslsc II** *T...$++=B  8 8 $"4"45  Q r)rN)rrrr rrr!rrrtorchnn Parameterdictstrr __classcell__)r,s@rr r %sj2jTX5=d>O>O5P$ UXX'' ((4UXX=O=O8O3P*Q Q r)r)abcrrcollections.abcrtypingrrr0torch.distributed distributedr2torch.distributed.algorithms.model_averaging.utils algorithmsmodel_averagingr.torch.utils._typing_utilsrr__all__r r r"rrr@sG#$" BB; 3 4"C"*]M]r