o j9:jU@sddlZddlZddlmZddlmZddlZddlmZddl m Z ddl m Z ddl mZddlmZmZmZgd ZeeZ d ZGd d d ZGd ddeZeedejZdZGdddZGdddZdede de!efddZ"dej#dede deej#fddZ$ddZ%  d)de&ed fd!e'e(efdBd"e d#e&ed fdBd$e'e(efdBde&e!e&e!e'ff d%d&Z)d"e!efd'd(Z*dS)*N)Sequence)Any)DTensor) local_map map_aggregate) BlockMask) tree_flattentree_maptree_unflatten)TensorChunkSpecsplit_args_kwargs_into_chunks merge_chunksFc@seZdZdZddZdS)_CustomReducera$ Custom reducer class that can be used to specify a custom operation that reduces losses of multiple microbatches into one value. Example: >>> # xdoctest: +SKIP >>> sum_reducer = _CustomReducer( >>> torch.tensor(0.0), >>> lambda a, b: a + b >>> ) cCs||_||_dSN) init_value reduce_fn)selfrrrn/home/nk/hobo-godmode/plappi-mvp/.venv/lib/python3.10/site-packages/torch/distributed/pipelining/microbatch.py__init__-s z_CustomReducer.__init__N)__name__ __module__ __qualname____doc__rrrrrr s rc@ eZdZdS) _LossReducerNrrrrrrrr2rgc@sfeZdZUdZddZeed<ddZddZe d e ed ffd d Z e d e e effd dZdS)r z2 Class used to specify chunking of inputs cCs ||_dSr split_dim)rr rrrrBs zTensorChunkSpec.__init__r cCs |jjd|jjd|jdS)N.()) __class__rrr rrrr__repr__GszTensorChunkSpec.__repr__cCsd|jdS)NzTensorChunkSpec(r#rr%rrr__str__LszTensorChunkSpec.__str__ chunk_dims.cCt|dd}|S)a A helper for creating a tuple of `TensorChunkSpec` from a tuple of chunk dimensions (int's). Example: >>> # xdoctest: +SKIP >>> # There are three positional arguments to the model, and >>> # we are chunking them along dimension 0, 0 and 1, respectively >>> args_chunk_spec = TensorChunkSpec.from_tuple((0, 0, 1)) cSt|Srr dimrrr^z,TensorChunkSpec.from_tuple..r)r(args_chunk_specrrr from_tupleOs  zTensorChunkSpec.from_tuplecCr))a\ A helper for creating a dictionary of `TensorChunkSpec` from a dictionary of chunk dimensions (int's). Example: >>> # xdoctest: +SKIP >>> # Chunk dimension 0 for the "id" argument, 1 for the "mask" argument >>> kwargs_chunk_spec = TensorChunkSpec.from_dict({"id": 0, "mask": 1}) cSr*rr+r,rrrr.pr/z+TensorChunkSpec.from_dict..r)r(kwargs_chunk_specrrr from_dictbs  zTensorChunkSpec.from_dictN)rrrrrint__annotations__r&r' staticmethodtupler1dictstrr3rrrrr =s    r c@r) _ReplicateNrrrrrr:vrr: block_mask num_chunksreturnc s jddkr g|Sjd|kstdd}tj||}tj||}jdur8tj||ndg|}jdurJtj||ndg|}g}d}t|D],} fdd} | t j || || || || j | |j d||| d7}qW|S)a Given a block mask, split the block mask along the batch dimension (dim0). Args: block_mask: Block mask to split num_chunks: Number of chunks to split the block mask into Returns: chunk_block_masks: List of chunked block masks rz;Block mask has fewer batch size than the number of chunks. Ncsfdd}|S)Ncs t|}|||||Sr)torch full_likemask_mod)bhq_idxkv_idxb_offset)r;idxrrbatch_offset_mask_mods zI_split_block_mask..create_mask_mod..batch_offset_mask_modr)rGrHr;)rGrcreate_mask_modsz*_split_block_mask..create_mask_mod) kv_num_blocks kv_indicesfull_kv_num_blocksfull_kv_indices BLOCK_SIZErA seq_lengths)rKsizeAssertionErrorr? tensor_splitrLrMrNrangeappendrfrom_kv_blocksrOrP) r;r< batch_dimkv_num_blocks_chunkskv_indices_chunksfull_kv_num_blocks_chunksfull_kv_indices_chunkschunk_block_masks batch_offset chunk_idxrJrrIr_split_block_maskzsH      r_tensorspecc s|jkstd|jdt|t}|r3|j}tfdd|f|fd}||}nt|j}|j rJ|j rJ|D]}| qCt sN|Sdtj dtj dttj d fffd d }|r|j}t|} t||f| |f|f| d} t| |g|RSt||g|RS) zGiven a tensor, and a chunking spec, split the tensor. Args: tensor: Tensor to split spec: Chunking spec num_chunks: Number of chunks to split the tensor into Returns: chunk_tensors: List of chunked tensors z Tensor size z is smaller than num_chunkscst|jSr)r?rSr )tr<rarrr.z_split_tensor..out_placements in_placementsorigchunksr=.csvg}d}|D]0}t|}||j}tdg|j}t|||j<|||<||||j7}qt|S)Nr)r? zeros_likerQr slicendimrUr7)rhriexpandedrGchunknew_valupperslices)rarr_expand_chunkss  z%_split_tensor.._expand_chunks)rQr rR isinstancer placementsrr?rS requires_gradis_leaf retain_grad_debug_mask_minibatchesTensorr7lenlist) r`rar< _is_dtensorrtsplit_fn chunk_tensorsrnrrn expand_fnrrcr _split_tensorsH      rcsP|s ddt|DSt|t|ks%tdt|dt||dur-tdt|ddd \}t|d dd \}}g}t||d d D]o\}}|tusWt|tr]| |qJt|t j r{t|t sqtd t || ||jqJt|trt|t std t ||jdkstd|jddkr| |qJ| |jdqJtd|d|dtg||R} ddt| D} t||d d D]K\}}g} |tust|tr|g| } n$t|t j rt||| } nt|trt|| } n td|d|dt| | d d D] \} } | | qqӇfdd| DS)aW Given a dictionary of args, and a dictionary of chunking specs, shard the args according to the chunking specs. Args: args_dict: Dictionary of args args_chunk_spec: Dictionary of chunking specs num_chunks: Number of chunks to shard the args into Returns: args_split: List of sharded args cSsg|]}iqSrr.0_rrr rdz'_shard_dict_of_args..zargs_dict.keys() = z args_chunk_spec.keys() = Nz.args_chunk_spec should have been set by callercS t|tSrrsrxrrrr.% z%_shard_dict_of_args..rvcSrrrrrrrr.(rTstrictzExpected TensorChunkSpec, got rz#BlockMask only supports split_dim=0r>zUnsupported chunk spec: z and value: z combination.cSsg|]}gqSrrrrrrrFrdcsg|]}t|qSr)r )r_flat_split_result tree_specrrrYs)rTrzrRr{keysr zipr:rsrUr?ryr typerQr rrK ValueErrorminrr_) args_dictr0r<values chunk_specsr split_sizesvraresult_num_chunksflat_split_resultsv_splitsr_v_splitrrr_shard_dict_of_argssp               rargs.kwargsrir0r2c Cs|duri}dd}|durt||ddd}|dur$t||ddd}ttt|tt||}t|}t|||}t||krTt|}ttt|tt||}t|t|krjtdt|d t|d d |D} | |fS) a Given a sequence of args and kwargs, split them into a number of chunks according to their respective chunking specs. Args: args: Tuple of args kwargs: Dict of kwargs chunks: Number of chunks to split the args and kwargs into args_chunk_spec: chunking specs for args, in same shape as args kwargs_chunk_spec: chunking specs for kwargs, in same shape as kwargs Returns: args_split: List of sharded args kwargs_split: List of sharded kwargs NcSst|tjtBr ttStSr)rsr?ryrr DEFAULT_CHUNK_DIMr:rrrr default_specsz3split_args_kwargs_into_chunks..default_speccSrrrrrrrr.rz/split_args_kwargs_into_chunks..rcSrrrrrrrr.rz;args and kwargs are split into different number of chunks: z, cs*g|]tfddttDqS)c3s|]}|VqdSrr)ri chunk_argsrr sz;split_args_kwargs_into_chunks...)r7rTrz)rrrrrsz1split_args_kwargs_into_chunks..)r rr8 enumeraterz RuntimeError) rrrir0r2rargs_split_dictreal_num_chunks kwargs_split args_splitrrrr _sR8       r c s:|dur t|\}}nt|d\}}ttgt|}g|D]}t|\}}t|t|kr:td|d||q g}t|D]P\ttrGfddttD} t r| dj } | ddD]} | j | ks~t d| d | j qlt j t j| d d it| jd } g} d}t| t| kst d t| dt| t| | ddD])\}}||j}tdddg|j}t|||j<||}| ||}qn| } dd| D}t|r;t|st d| djt| dddD]\}}|jkrt d|dd |jqtfddftfddtt| Dd}||| qF|t j| jdqFttrjj}ttD] }||}qV||qFd}tdtD]}||kst d|d |qw||qFt||S)z Given a list of chunks, merge them into a single value according to the chunk spec. Args: chunks: list of chunks chunk_spec: Chunking spec for the chunks Returns: value: Merged value NrzChunk z did not match chunk spec csg|]}|qSrr)rr^)arg_idxchunks_flattenedrrrs z merge_chunks..r>zExpected shape z, got devicemeta)sectionsr-z6Expected len(partial_values) == len(meta_chunks), got z != TrcSsg|]}t|tqSr)rsr)rrrrrrCszRmerge_chunks: expected all values to be DTensors or none to be DTensors, got a mixz*merge_chunks: placement mismatch at chunk z : expected cstj|jdS)Nr,)r?catr )ri)argrrr.Srdzmerge_chunks..c3s|]}VqdSrrr)rtrrrUszmerge_chunks..rer,z Expected )r r rrzrrUrrsrTrxshaperRr?rSemptyr rrQrkrlanyallrtrr7rrrrr )ri chunk_specspec_flattened flatten_spec chunk0_flatrnchunk_flattenedrargs_flattenedpartial_values overall_shapeval meta_chunks values_to_catchunk_start_idx partial_value meta_chunk chunk_end_idx slice_indicessliced dtensor_flagsrrcat_fn reduced_valr^valuer)rrrrtrrs-                   r)NN)+loggingoperatorcollections.abcrtypingrr?torch.distributed.tensorr%torch.distributed.tensor.experimentalr torch.fx.noder!torch.nn.attention.flex_attentionrtorch.utils._pytreer r r __all__ getLoggerrloggerrxrrr`add sum_reducerrr r:r4r{r_ryrrr7r8r9r rrrrrsn       9 A L\  s