o h9@sUdZddlZddlmZmZmZmZmZmZddlZddlm Z ddl ZgdZ ee e d<ee Zee efZeegeeee ffZGdddZGd d d Zdd d Zded eefddZdeefddZdS)zManipulation of micro-batches.N)AnyCallableListUnioncastSequence)Tensor)NoChunkBatchcheckscattergather__all__c@s*eZdZdZdefddZeddZdS)r a 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|s td|||_dS)Nz+NoChunk only supported for tensors, found: )torch is_tensor TypeError_tensor)selfrr`/var/www/html/ai/venv/lib/python3.10/site-packages/torch/distributed/pipeline/sync/microbatch.py__init__s  zNoChunk.__init__cC|jSN)rrrrrtensor#szNoChunk.tensorN)__name__ __module__ __qualname____doc__rrpropertyrrrrrr s r c@seZdZdZdeeeefddfddZe defddZ e d d Z d d Z d dZ deddfddZdefddZddZdefddZdefddZejdededdfddZejdededdfddZdeeefddfddZdeddfd d!Zdeddfd"d#ZdS)$r zC An abstraction representing a microbatch in the pipeline. valuesreturnNcCsD||_t||_|jstdd|jDs td|jdSdS)Ncs|]}t|VqdSrrr).0valuerrr 3z!Batch.__init__..zNo tensors found in batch: )_valuesrratomicanyr)rr!rrrr-s zBatch.__init__cCs|jstdtt|jS)z Retrieves the underlying tensor.znot atomic batch)r*AttributeErrorrrr)rrrrr6s z Batch.tensorcCr)z-Retrieves the underlying values for the batch)r)rrrrr!=sz Batch.valuescCs8|jrdSt|jD] \}}t|r|Sq td)z< Retrieves the index of first tensor found. rzNo tensor found!)r* enumerater)rrr)rir&rrrfind_tensor_idxBs zBatch.find_tensor_idxcCs2|jr|jjS|jD] }t|r|jSq dS)z; Retrieves the device for this microbatch. N)r*r)devicerr)rr&rrr get_deviceNs   zBatch.get_devicefunctioncCs"|jr t||jSt||jS)zbCalls a function on the microbatch. It also wraps the output with :class:`Batch`. )r*r r))rr2rrrcallYsz Batch.callcCsd|jd|jdS)Nz Batch[atomic=z]()r*r)rrrr__repr__bszBatch.__repr__ccs$|jr |jVdS|jEdHdSrr5rrrr__iter__es zBatch.__iter__cCs|jrdSt|jS)N)r*lenr)rrrr__len__ksz Batch.__len__indexcCs&|js|j|S|dkrtd|jS)Nr atomic batch allows index 0 onlyr*r) IndexError)rr;rrr __getitem__ns  zBatch.__getitem__r&cCdSrrrr;r&rrr __setitem__xzBatch.__setitem__cCr@rrrArrrrB|rCcCs*t|tr |||dS|||dSr) isinstanceint_setitem_by_index_setitem_by_slicerArrrrBs cCsP|js|}|jd||f|j|dd|_dS|dkr#td||_dS)Nr8rr<r=)rr;r&r.rrrrFs( zBatch._setitem_by_indexcCsf|j|jur|jurdustdtd|js"||_dSt|dkr,td|d|_dS)Nzonly slice [:] supportedr8z5atomic batch cannot be replaced with multiple tensorsr)startstopstepNotImplementedErrorr*r)r9r>rArrrrGs" zBatch._setitem_by_slice)rrrrrrrrrr rr!r/r1Functionr3strr6r7rEr:r?typingoverloadrBsliceTensorsrFrGrrrrr (s*       r r"csBtdd|Dstd|tfdd|DrtddS)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 csr#rr$r%inputrrrr'r(zcheck..z inputs do not have any tensors: c3s$|] }t|o |jkVqdSr)rrr0rR first_devicerrr's"z>All inputs should be on the same device as the first partitionN)r+r ValueError)rUinputsrrTrr s  r chunkscGst|dkrt|dtrdd|d|DSddt|D}d}|D]S}t|r]||}|dkrH|t|krHtd|dt|t|}t|D] \}}|| |qPq&t|D]}t|t rq|| |j qa|| |qaq&|d |}d d|DS) z7Splits an input mini-batch into multiple micro-batches.r8rcSg|]}t|qSrr r%xrrr zscatter..cSsg|]}gqSrr)r%_rrrr]sz6Found different number of chunks produced for inputs:  and NcSrYrrZr[rrrr]r^) r9rDrchunkrangerr RuntimeErrorr-appendr r)rXrWbatches num_chunksrStensorsr.rrrrr s(     r outputsc Cs|djrtdd|D}t|}|Sg}tt|dD]E}t|d|}g}|D]}|t||krEtd|dt|||||q-t |d|r_|t|q||qt|}|S)z4Concatenates output micro-batches into a mini-batch.rcss|]}|jVqdSr)r)r%brrrr'szgather..z2Types for microbatch outputs do not match, found: ra) r*tuplercatrcr9typerrer)rirhoutput output_bufr. output_typecurrent_outputsbatchrrrr s"   r )r"N)rrNrrrrrrrrtorch.cuda.commrrM__annotations__rQTensorOrTensorsrLr r r rEr r rrrrs    w#