U d,@sdZddlmZddlmZddlZddlmZmZm Z m Z m Z m Z m Z mZmZddlZddlmZddlZddlmZmZdd lmZdd lmZd d gZeeZe eefZeeefZeee efZerdd lm Z ne!Z Gddde Z"e"dddZ#GdddZ$Gdddej%Z&e&Z'ee ddddZ(ee ddddZ)e*ddd Z+e*ddd Z,Gdd d Z-ej.e edd!d"d#Z/eej.e ee dd!d$d%Z0Gd&d'd'ej1j"Z2Gd(d)d)ej1j"Z3dS)*aCheckpointing with preceding recomputation. PyTorch already provides the official checkpointing utilities in :mod:`torch.utils.checkpoint`. The official checkpointing combines recomputation and recursive backpropagation into one autograd function named ``CheckpointFunction``. Hence, the recomputation can be started only when the gradients arrive to the function. In Pipe, the recomputation needs to precede the gradient arrival to minimize the GPU idle time. We solve this problem by introducing separate autograd functions named :class:`Recompute` and :class:`Checkpoint`. Each function represents recomputation and recursive backpropagation, respectively. We can manipulate the control flow in aspect of both the autograd engine and CUDA with a pair of the functions. Specifically, we place CUDA stream synchronization between :class:`Recompute` and :class:`Checkpoint` to delay only :class:`Checkpoint` until the gradient is copied entirely. )deque)contextmanagerN) TYPE_CHECKINGAnyDeque GeneratorListOptionalUnionSequenceTuple)Tensor)forkjoin)Batch) get_phonyis_checkpointingis_recomputing)Protocolc@seZdZeedddZdS)Function)inputreturncCsdSN)selfrrrN/tmp/pip-unpacked-wheel-ua33x9lu/torch/distributed/pipeline/sync/checkpoint.py__call__FszFunction.__call__N)__name__ __module__ __qualname__TensorOrTensorsrrrrrrEsr)functioncCs*t|}t||}|}|||jS)zMakes a checkpoint with a simple interface like :func:`torch.utils.checkpoint.checkpoint`. It's only used to test or debug :class:`Checkpoint` and :class:`Recompute` without boilerplate. )r Checkpointing checkpoint recomputevalues)r"rbatchZchkrrrr$Js   r$c@s@eZdZdZeeddddZedddZedd d d ZdS) r#z?Generates a pair of :class:`Checkpoint` and :class:`Recompute`.N)r"r'rcCs(||_||_tdd|_tdd|_dS)Nr)maxlen)r"r'r recomputed rng_states)rr"r'rrr__init__[s zCheckpointing.__init__rcCsf|jj}t|j}t|jdd}tj||j|j|j |f|}t |tr^tdd|D}t |S)z/Returns a batch applied by :class:`Checkpoint`.T) requires_gradcSs*g|]"}t|r"|s"|n|qSr)torch is_tensorZis_floating_pointdetach.0xrrr rsz,Checkpointing.checkpoint..) r'atomictuplerZ get_device Checkpointapplyr)r*r" isinstancer)r input_atomicinputsphonyoutputrrrr$ds  zCheckpointing.checkpoint)r'rcCsb|jj}t|j}|}t||\||<}tj||j|j|j |f|}t |||||<dS)z1Applies :class:`Recompute` to the batch in place.N) r'r5r6Zfind_tensor_idxr Recomputer8r)r*r"r)rr'r:r;Z tensor_idxr<rrrr%vs  zCheckpointing.recompute) rrr __doc__rrr+r$r%rrrrr#Xs r#c@seZdZddddZdS) ThreadLocalNr,cCsd|_d|_dS)NF)rr)rrrrr+szThreadLocal.__init__)rrr r+rrrrr@sr@)NNNr,ccs$tj}dt_z dVW5|t_XdS)zDMakes :func:`is_checkpointing` return :data:`True` within a context.TN thread_localrorigrrrenable_checkpointings  rEccs$tj}dt_z dVW5|t_XdS)zBMakes :func:`is_recomputing` return :data:`True` within a context.TNrBrrCrrrenable_recomputings  rGcCstjS)zWhether the current forward propagation is under checkpointing. Returns: bool: :data:`True` if it's under checkpointing. rArrrrrscCstjS)a9Whether the current forward propagation is under checkpoint recomputation. Use this to prevent duplicated side-effects at forward propagation:: class Counter(nn.Module): def __init__(self): super().__init__() self.counter = 0 def forward(self, input): if not is_recomputing(): self.counter += 1 return input Returns: bool: :data:`True` if it's under checkpoint recomputation. .. seealso:: :ref:`Detecting Recomputation` rFrrrrrsc@sfeZdZUdZeeed<eeed<eed<e ed<e e ed<e e dfed<e d d d d Zd S) Contextz]The common interface between the :class:`Checkpoint` and :class:`Recompute` context. r)r*r"r:r;.Z saved_tensorsN)tensorsrcGsdSrr)rrIrrrsave_for_backwardszContext.save_for_backward)rrr r?r Recomputed__annotations__ RNGStatesrboolr rr r rJrrrrrHs    rH)devicer*rcCs6t}|jdkr tj|}nd}|||fdS)z:meth:`Checkpoint.forward` captures the current PyTorch's random number generator states at CPU and GPU to reuse in :meth:`Recompute.backward`. .. seealso:: :ref:`Referential Transparency` cudaN)r.Z get_rng_statetyperPappend)rOr* cpu_rng_state gpu_rng_staterrrsave_rng_statess  rUc csf|\}}g}|jdkr$||tj|,t||dk rRtj||dVW5QRXdS)z:meth:`Recompute.backward` restores the random number generator states captured by :func:`save_rng_states` within its context. .. seealso:: :ref:`Referential Transparency` rPN)poprQrRr.randomZfork_rngZ set_rng_staterP)rOr*rSrTZ gpu_devicesrrrrestore_rng_statess    rXc@sReZdZeeeeeeee e dddZ eeee e edfdddZdS) r7)ctxr<r)r*r"r:c Gs||_||_t|j|j||_||_|r6|dg}n"g}|D]}t|r>||q>|j |t Bt 0|rt |dkst ||d} n||} W5QRXW5QRX| S)Nrr)r)r*rUrOr"r:r.r/rRrJZno_gradrElenAssertionError) rYr<r)r*r"r:r;rIrr=rrrforwards$     zCheckpoint.forward.rY grad_outputrcGs|j\}}t|tr|}n|f}tdd|DrVtdd|D}tj||dddddg}|dd|Dt|S)Ncss|]}t|o|jVqdSrr.r/r-)r2yrrr *sz&Checkpoint.backward..cSs g|]}t|r|jr|qSrr_r1rrrr4+s z'Checkpoint.backward..css"|]}t|r|jndVqdSr)r.r/Zgradr1rrrra/s) r)rVr9r6anyr.autogradbackwardextend)rYr^r=Z input_leafoutputsrI grad_inputrrrrd"s zCheckpoint.backwardN)rrr staticmethodrHr rrKrMrrNr\r r rdrrrrr7s"r7c @sLeZdZeeeeeeee e edddZ eeee ddddZ dS) r>)rYr<r)r*r"r:rc Gs^||_||_||_||_||_|r.|dg}n"g}|D]}t|r6||q6|j||S)Nr) r)r*r"r:r;r.r/rRrJ) rYr<r)r*r"r:r;rIrrrrr\4s     zRecompute.forward)N.r]c Gs|j}tdd|D}d}|D]}t|r |j}q:q |dkrPtd|t||j\tHt 6|j rt |dkst | |d}n |j |}W5QRXW5QRXW5QRX|j||fdddddg}|dd|jDt|S)Ncss,|]$}t|r ||jn|VqdSr)r.r/r0Zrequires_grad_r-r1rrrraSsz%Recompute.backward..zNo tensors found in rrcss|] }dVqdSrr)r2_rrrrajs)r;r6r.r/rO RuntimeErrorrXr*Z enable_gradrGr:rZr[r"r)rRre)rYr^r;Z inputs_leafrOrr=rgrrrrdPs& (zRecompute.backwardN)rrr rhrHr rrKrMrrNr\r rdrrrrr>3sr>)4r? collectionsr contextlibr threadingtypingrrrrrr r r r r.r Ztorch.autograd dependencyrrZ microbatchrr<r__all__ZTensorsr!rKrMZtyping_extensionsrobjectrr$r#localr@rBrErGrNrrrHrOrUrXrcr7r>rrrrsF  ,      ,   6