U d@sUdZddlZddlmZmZmZmZmZmZddlZddlm Z ddl ZgZ ee e d<ee Zee efZeegeeee ffZGdddeZGdd d Zdd d d Zeeed ddZeedddZdS)zManipulation of micro-batches.N)AnyCallableListUnioncastSequence)Tensor__all__c@s*eZdZdZedddZeddZdS)NoChunka Wrapper for a Tensor in :meth:`Pipe.forward` indicating that the tensor should not be chunked on the batch dimension and instead be replicated as-is across all micro-batches. This is useful for tensors which might not have any 'batch' semantics for the model. )inpcCs"t|std|||_dS)Nz+NoChunk only supported for tensors, found: )torch is_tensor TypeError_tensor)selfr rN/tmp/pip-unpacked-wheel-ua33x9lu/torch/distributed/pipeline/sync/microbatch.py__init__s zNoChunk.__init__cCs|jSN)rrrrrtensor#szNoChunk.tensorN)__name__ __module__ __qualname____doc__rrpropertyrrrrrr sr c@seZdZdZeeeefddddZe edddZ e d d Z d d Z d dZ eddddZedddZddZedddZedddZejeeddddZejeeddddZeeefddd dZeddd!d"Zeddd#d$ZdS)%BatchzC An abstraction representing a microbatch in the pipeline. N)valuesreturncCs@||_t||_|js3sz!Batch.__init__..zNo tensors found in batch: )_valuesr r atomicanyr)rrrrrr-s  zBatch.__init__rcCs|jstdtt|jS)z Retrieves the underlying tensor.znot atomic batch)r$AttributeErrorrrr#rrrrr6sz Batch.tensorcCs|jS)z-Retreives the underlying values for the batch)r#rrrrr=sz Batch.valuescCs<|jr dSt|jD]\}}t|r|SqtddS)z< Retrieves the index of first tensor found. rzNo tensor found!N)r$ enumerater#r r r)rir!rrrfind_tensor_idxBs   zBatch.find_tensor_idxcCs2|jr|jjS|jD]}t|r|jSqdS)z; Retrieves the device for this microbatch. N)r$r#devicer r )rr!rrr get_deviceNs   zBatch.get_device)functionrcCs&|jrt||jSt||jSdS)zbCalls a function on the microbatch. It also wraps the output with :class:`Batch`. N)r$rr#)rr-rrrcallYsz Batch.callcCsd|jd|jdS)Nz Batch[atomic=z]()r$r#rrrr__repr__bszBatch.__repr__ccs |jr|jVn |jEdHdSrr0rrrr__iter__es zBatch.__iter__cCs|jr dSt|jS)N)r$lenr#rrrr__len__ksz Batch.__len__)indexcCs&|js|j|S|dkr td|jS)Nr atomic batch allows index 0 onlyr$r# IndexError)rr6rrr __getitem__ns  zBatch.__getitem__)r6r!rcCsdSrrrr6r!rrr __setitem__xszBatch.__setitem__cCsdSrrr;rrrr<|s)r6rcCs(t|tr|||n |||dSr) isinstanceint_setitem_by_index_setitem_by_slicer;rrrr<s cCsP|js6|}|jd||f|j|dd|_dS|dkrFtd||_dS)Nr3rr7r8)rr6r!r)rrrr?s(zBatch._setitem_by_indexcCs`|j|jkr$|jkr$dks.ntd|js>||_dSt|dkrRtd|d|_dS)Nzonly slice [:] supportedr3z5atomic batch cannot be replaced with multiple tensorsr)startstopstepNotImplementedErrorr$r#r4r9r;rrrr@s& zBatch._setitem_by_slice)rrrrrrrrrrrrr*r,Functionr.strr1r2r>r5r:typingoverloadr<sliceTensorsr?r@rrrrr(s(       rr&csBtdd|Ds td|tfdd|Dr>tddS)z Checks whether the input contains at least one tensor and each tensor is on the same device as the first partition. Raises: ValueError: input does not contain at least one tensor css|]}t|VqdSrrr inputrrrr"szcheck..z inputs do not have any tensors: c3s"|]}t|o|jkVqdSr)r r r+rK first_devicerrr"sz>All inputs should be on the same device as the first partitionN)r%r ValueError)rNinputsrrMrchecks rQ)chunksrcGst|dkr2t|dtr2dd|d|DSddt|D}d}|D]}t|r||}|dkr|t|krtd|dt|t|}t|D]\}}|| |qqLt|D].}t|t r|| |j q|| |qqL|d |}d d|DS) z7Splits an input mini-batch into multiple micro-batches.r3rcSsg|] }t|qSrrr xrrr szscatter..cSsg|]}gqSrr)r _rrrrVsz6Found different number of chunks produced for inputs:  and NcSsg|] }t|qSrrSrTrrrrVs) r4r=rchunkranger r RuntimeErrorr(appendr r)rRrPZbatchesZ num_chunksrLtensorsr)rrrrscatters$     r_)outputsc Cs|djr(tdd|D}t|}ng}tt|dD]}t|d|}g}|D]>}|t||krtd|dt|||||qXt |d|r|t|q<||q.z2Types for microbatch outputs do not match, found: rY) r$tupler catr[r4typerr]r )r`r^outputZ output_bufr) output_typeZcurrent_outputsbatchrrrgathers    rh)rrGrrrrrrr rZtorch.cuda.commr rF__annotations__rJZTensorOrTensorsrEobjectr rrQr>r_rhrrrrs   w#