o i[F@sddlZddlZddlmZddlmZddlZddlmZddl m Z ddl m Z m Z mZgdZeeZ dZGd d d ZGd d d eZeed ejZdZGdddZGdddZde dedee fddZdejdededeejfddZ ddZ!  d'de"edfde#e$efdBd ed!e"edfdBd"e#e$efdBde"ee"ee#ff d#d$Z%d eefd%d&Z&dS)(N)Sequence)Any 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)selfrrri/sda-disk/www/egybert/egybert_env/lib/python3.10/site-packages/torch/distributed/pipelining/microbatch.py__init__+s z_CustomReducer.__init__N)__name__ __module__ __qualname____doc__rrrrrr s r c@ eZdZdS) _LossReducerNrrrrrrrr0rgc@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)rrrrrr@s zTensorChunkSpec.__init__rcCs |jjd|jjd|jdS)N.()) __class__rrrrrrr__repr__EszTensorChunkSpec.__repr__cCsd|jdS)NzTensorChunkSpec(r!rr#rrr__str__JszTensorChunkSpec.__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_tupleMs  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,nr-z+TensorChunkSpec.from_dict..r)r&kwargs_chunk_specrrr from_dict`s  zTensorChunkSpec.from_dictN)rrrrrint__annotations__r$r% staticmethodtupler/dictstrr1rrrrr ;s    r c@r) _ReplicateNrrrrrr8trr8 block_mask num_chunksreturnc s jddkr g|Sjd|ksJdd}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)r9idxrrbatch_offset_mask_mods zI_split_block_mask..create_mask_mod..batch_offset_mask_modr)rErFr9)rErcreate_mask_modsz*_split_block_mask..create_mask_mod) kv_num_blocks kv_indicesfull_kv_num_blocksfull_kv_indices BLOCK_SIZEr? seq_lengths) rIsizer= tensor_splitrJrKrLrangeappendrfrom_kv_blocksrMrN) r9r: batch_dimkv_num_blocks_chunkskv_indices_chunksfull_kv_num_blocks_chunksfull_kv_indices_chunkschunk_block_masks batch_offset chunk_idxrHrrGr_split_block_maskxsF      r\tensorspecc Cs||j|ksJd||jdt|||j}ts |Sg}d}|D]2}t|}|||j}tdddg|j} t||| |j<||| <|||||j7}q&|S)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_chunksrN) rOrr=rP_debug_mask_minibatches zeros_likeslicendimrR) r]r^r: chunk_tensorsexpanded_chunks split_dim_idx chunk_tensornew_val upper_idx slice_indicesrrr _split_tensors"  rjcs.|s ddt|DSt|t|ks%Jdt|dt||dus+Jt|ddd\}t|d dd\}}g}t||d d D]a\}}|tusUt|tr[||qHt|t j rrt|t shJ|| |j qHt|trt|t s~J|j d ksJd |j d dkr||qH||j d qHtd|d|dtg||R} ddt| D} t||d d D]J\}}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 z'_shard_dict_of_args..zargs_dict.keys() = z args_chunk_spec.keys() = NcS t|tSr isinstancerxrrrr, z%_shard_dict_of_args..is_leafcSrprrqrsrrrr,ruTstrictrz#BlockMask only supports split_dim=0r<zUnsupported chunk spec: z and value: z combination.cSsg|]}gqSrrrkrrrrnrocsg|]}t|qSr)r )rl_flat_split_result tree_specrrrn.s)rQlenlistkeysrzipr8rrrRr=Tensorr rOrrrI ValueErrorminrjr\) args_dictr.r:values chunk_specsrm split_sizesvr^result_num_chunksflat_split_resultsv_splitsrz_v_splitrr{r_shard_dict_of_argssf             rargs.kwargschunksr.r0c 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)rrr=rrr DEFAULT_CHUNK_DIMr8rrrr default_specqsz3split_args_kwargs_into_chunks..default_speccSrprrqrrrrr,yruz/split_args_kwargs_into_chunks..rvcSrprrqrrrrr,~ruz;args and kwargs are split into different number of chunks: z, cs*g|]tfddttDqS)c3s|]}|VqdSrr)rli chunk_argsrr sz;split_args_kwargs_into_chunks...)r5rQr})rlrrrrnsz1split_args_kwargs_into_chunks..)rrr6 enumerater} RuntimeError) rrrr.r0rargs_split_dictreal_num_chunks kwargs_split args_splitrrrr 4sR8       r cs6|dur t|\}}nt|d\}}ttgt|}g|D]}t|\}}t|t|kr:td|d||q g}t|D]\} t| trӇfddttD} t r| dj } | ddD] } | j | kssJqjt j t j | dd it| | jd } g}d}t| t| ksJt| | d d D])\}}||| j}tdddg|j}t|||| j<||}|||}qn| }|t j|| jd qFt| tr| j}ttD] }| ||}q||qFd}tdtD]}||ksJq||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)rlr[arg_idxchunks_flattenedrrrns z merge_chunks..r<devicemeta)sectionsr+Trxr*)rr rr}rrRrrrrQr_shaper=rPemptyrrrOrarbcatr rrr )r chunk_specspec_flattened flatten_spec chunk0_flatchunkchunk_flattenedrmargs_flattenedargpartial_values overall_shapeval meta_chunks values_to_catchunk_start_idx partial_value meta_chunk chunk_end_idxrisliced reduced_valr[valuerrrr sh-              r )NN)'loggingoperatorcollections.abcrtypingrr= torch.fx.noder!torch.nn.attention.flex_attentionrtorch.utils._pytreerrr __all__ getLoggerrloggerr_r rr]add sum_reducerrr r8r2r~r\rrjrr5r6r7r r rrrrsj     9 @ )W  s