U ʹj5- @sdZddlmZddlmZddlZddlmZmZm Z m Z m Z m Z m Z mZmZddlZddlmZddlZddlmZmZdd lmZdd lmZd d d ddddddddddg ZeeZe eefZeeefZeee efZGdd d e Zeddd Z Gdd d Z!Gdddej"Z#e#Z$ee ddddZ%ee ddd dZ&e'dd!dZ(e'dd"dZ)Gd#ddZ*ej+eedd$d%dZ,eej+eee dd$d&dZ-Gd'ddej.jZ/Gd(ddej.jZ0dS))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) AnyDeque GeneratorListOptionalProtocolUnionSequenceTuple)Tensor)forkjoin)Batch) get_phonyFunction checkpoint Checkpointing ThreadLocalenable_checkpointingenable_recomputingis_checkpointingis_recomputingContextsave_rng_statesrestore_rng_states Checkpoint Recomputec@seZdZeedddZdS)r)inputreturncCsdSN)selfr r#r#i/var/www/html/bwave/python/venv/lib/python3.8/site-packages/torch/distributed/pipeline/sync/checkpoint.py__call__BszFunction.__call__N)__name__ __module__ __qualname__TensorOrTensorsr&r#r#r#r%rAs)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. )rrr recomputevalues)r+r batchZchkr#r#r%rFs   c@s@eZdZdZeeddddZedddZedd d d ZdS) rz?Generates a pair of :class:`Checkpoint` and :class:`Recompute`.N)r+r.r!cCs(||_||_tdd|_tdd|_dS)Nr)maxlen)r+r.r recomputed rng_states)r$r+r.r#r#r%__init__Ws zCheckpointing.__init__r!cCsf|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.0xr#r#r% nsz,Checkpointing.checkpoint..) r.atomictuplerZ get_devicerapplyr0r1r+ isinstancer)r$ input_atomicinputsphonyoutputr#r#r%r`s  zCheckpointing.checkpoint)r.r!cCsb|jj}t|j}|}t||\||<}tj||j|j|j |f|}t |||||<dS)z1Applies :class:`Recompute` to the batch in place.N) r.r<r=Zfind_tensor_idxrrr>r0r1r+r)r$r.r@rAZ tensor_idxrBr#r#r%r,rs  zCheckpointing.recompute) r'r(r)__doc__rrr2rr,r#r#r#r%rTs c@seZdZddddZdS)rNr3cCsd|_d|_dS)NF)rr)r$r#r#r%r2szThreadLocal.__init__)r'r(r)r2r#r#r#r%rs)NNNr3ccs$tj}dt_z dVW5|t_XdS)zDMakes :func:`is_checkpointing` return :data:`True` within a context.TN thread_localrorigr#r#r%rs  ccs$tj}dt_z dVW5|t_XdS)zBMakes :func:`is_recomputing` return :data:`True` within a context.TNrFrrGr#r#r%rs  cCstjS)zWhether the current forward propagation is under checkpointing. Returns: bool: :data:`True` if it's under checkpointing. rEr#r#r#r%rscCstjS)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` rIr#r#r#r%rsc@sfeZdZUdZeeed<eeed<eed<e ed<e e ed<e e dfed<e d d d d Zd S) rz]The common interface between the :class:`Checkpoint` and :class:`Recompute` context. r0r1r+r@rA.Z saved_tensorsN)tensorsr!cGsdSr"r#)r$rJr#r#r%save_for_backwardszContext.save_for_backward)r'r(r)rDr Recomputed__annotations__ RNGStatesrboolr rr r rKr#r#r#r%rs    )devicer1r!cCs6t}|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)r5Z get_rng_statetyperQappend)rPr1 cpu_rng_state gpu_rng_stater#r#r%rs  c 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` rQN)poprRrSr5randomZfork_rngZ set_rng_staterQ)rPr1rTrUZ gpu_devicesr#r#r%rs    c@sReZdZeeeeeeee e dddZ eeee e edfdddZdS) r)ctxrBr0r1r+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)r0r1rrPr+r@r5r6rSrKZno_gradrlenAssertionError) rXrBr0r1r+r@rArJr rCr#r#r%forwards$     zCheckpoint.forward.rX grad_outputr!cGs|j\}}t|tr|}n|f}tdd|DrVtdd|D}tj||dddddg}|dd|Dt|S)Ncss|]}t|o|jVqdSr"r5r6r4)r9yr#r#r% &sz&Checkpoint.backward..cSs g|]}t|r|jr|qSr#r^r8r#r#r%r;'s z'Checkpoint.backward..css"|]}t|r|jndVqdSr")r5r6Zgradr8r#r#r%r`+s) r0rVr?r=anyr5autogradbackwardextend)rXr]rCZ input_leafoutputsrJ grad_inputr#r#r%rcs zCheckpoint.backwardN)r'r(r) staticmethodrr rrLrNrrOr[r rrcr#r#r#r%rs"c @sLeZdZeeeeeeee e edddZ eeee ddddZ dS) r)rXrBr0r1r+r@r!c Gs^||_||_||_||_||_|r.|dg}n"g}|D]}t|r6||q6|j||S)Nr) r0r1r+r@rAr5r6rSrK) rXrBr0r1r+r@rArJr r#r#r%r[0s     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")r5r6r7Zrequires_grad_r4r8r#r#r%r`Osz%Recompute.backward..zNo tensors found in rrcss|] }dVqdSr"r#)r9_r#r#r%r`fs)rAr=r5r6rP RuntimeErrorrr1Z enable_gradrr@rYrZr+r0rSrd)rXr]rAZ inputs_leafrPr rCrfr#r#r%rcLs& (zRecompute.backwardN)r'r(r)rgrr rrLrNrrOr[r rcr#r#r#r%r/s)1rD collectionsr contextlibr threadingtypingrrrrrr r r r r5r Ztorch.autograd dependencyrrZ microbatchrrBr__all__ZTensorsr*rLrNrrrlocalrrFrrrOrrrrPrrrbrrr#r#r#r%sR  ,       ,   6