U yhA? @s&ddlZddlmZmZmZmZmZddlZddlm Z ddl m Z m Z dddgZ eeZdZGd d d ZGd d d eZeed ddZdZGdddZGdddZddZdeedfeeeefeeeedfeeeefeeeeefdddZeedddZdS)N)AnyDictListOptionalTuple map_aggregate) tree_flattentree_unflattenTensorChunkSpecsplit_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)selfrrrY/var/www/html/venv/lib/python3.8/site-packages/torch/distributed/pipelining/microbatch.py__init__(sz_CustomReducer.__init__N)__name__ __module__ __qualname____doc__rrrrrrs rc@s eZdZdS) _LossReducerNrrrrrrrr-srgcCs||Srr)abrrr1rc@sfeZdZUdZddZeed<ddZddZe e ed fd d d Z e e e efd d dZdS)r z2 Class used to specify chunking of inputs cCs ||_dSr split_dim)rr!rrrr=szTensorChunkSpec.__init__r!cCs |jjd|jjd|jdS)N.()) __class__rrr!rrrr__repr__BszTensorChunkSpec.__repr__cCsd|jdS)NzTensorChunkSpec(r$r r&rrr__str__GszTensorChunkSpec.__str__.) chunk_dimscCst|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)) cSst|Srr dimrrrrYrz,TensorChunkSpec.from_tuple..r)r)args_chunk_specrrr from_tupleJs zTensorChunkSpec.from_tuplecCst|dd}|S)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}) cSst|Srr*r+rrrrkrz+TensorChunkSpec.from_dict..r)r)kwargs_chunk_specrrr from_dict]s zTensorChunkSpec.from_dictN)rrrrrint__annotations__r'r( staticmethodrr.rstrr0rrrrr 8s   c@s eZdZdS) _ReplicateNrrrrrr5qsr5c# Csi}g}|}d}t|t|ksDtdt|dt||D]\}}t|\} } || ||} | dk stt| \} } t| t| krtd|d| g}t| | D]f\}}|t kst |t j s||g|qt |t rt |t j st|d||j}||krr|rVtd|d |d |d |}ntd |d |d|dt |||j}trg}d}|D]f}t |}|||j}tdddg|j}t||||j<|||<|||||j7}q||n ||d}qtd|q|||<qLg}t|D]N}i}|D]0\}}g}|D]}|||q^|||<qN||q>g}|D]V} i}!t|t| kstt| |D]\\}}}"t||"|!|<q||!q|S)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 Tzargs_dict.keys() = z args_chunk_spec.keys() = NzArgument value z9 did not have the same number of values as as chunk spec z is not a tensorz%Tensor size on chunking dimension is z', downsizing the number of chunks from z to r"zArg z% on chunking dimension has a size of z$, smaller than the number of chunks z. PiPPy cannot reduce the number of chunks because other arguments have bigger chunk-dimension sizes. Please adjust your num_chunks setting.rFzUnrecognized chunk spec: )lenAssertionErrorlistkeysitemsr append ValueErrorzipr5 isinstancetorchZTensorr sizer!loggerwarning RuntimeError tensor_split_debug_mask_minibatchesZ zeros_likeslicendim TypeErrorranger )#Z args_dictr-Z num_chunksZargs_sharded_replicatedZ arg_specsreal_num_chunksZ first_tensorZarg_keyargZflatspec chunk_specZchunk_spec_flat_Zsharded_arg_flatvZchunk_vZv_split_dim_sizeZ chunk_tensorsZexpanded_chunksZ split_dim_idxZ chunk_tensornew_valZ upper_idx slice_indicesZ chunks_flat chunk_idx chunk_argskeyZarg_single_chunkZv_flat args_splitchunkZper_chunk_argsZarg_specrrr_shard_dict_of_argsus               rW.)argskwargschunksr-r/returnc s|dkr i}|dkr&ttft|}|dkr>t|tt}ttt|tt||}t|}t|||}t||krt|}ttt|tt||}t|t|krtdt|dt|g}|D](|t fddt tDq||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 Nz;args and kwargs are split into different number of chunks: z, c3s|]}|VqdSrr).0irSrr Ysz0split_args_kwargs_into_chunks..) r DEFAULT_CHUNK_DIMr6dictfromkeysrW enumeraterCr;tuplerI) rXrYrZr-r/Zargs_split_dictrJZ kwargs_splitrUrr^rr s@8     &)rZcs@|dk rt|\}}n"t|d\}}ttgt|}g|D]>}t|\}}t|t|krttd|d||q@g}t|D]\} t| trfddttD} t r| dj } | ddD]} | j | kst qt j t j| dd it| | jd } g}d}t| t| ks.t t| | D]T\}}||| j}tdddg|j}t|||| j<||}|||}q8n| }|t j|| jd qt| tr| j}ttD]}| ||}q||qd}tdtD]}||ks t q ||qt||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)r\rRZarg_idxZchunks_flattenedrr sz merge_chunks..Zdevicemeta)sectionsr,r+)r r r`r6r<r;rcr>rIrEshaper7r?rDemptyr!r=r@rFrGcatrrrr )rZrMZspec_flattenedZ flatten_specZ chunk0_flatrVZchunk_flattenedrNZargs_flattenedrKZpartial_valuesZ overall_shapevalZ meta_chunksZ values_to_catZchunk_start_idxZ partial_valueZ meta_chunkZ chunk_end_idxrQZslicedZ reduced_valrRvaluerrerr ^sd-             )NN)loggingtypingrrrrrr?Z torch.fx.noderZtorch.utils._pytreer r __all__ getLoggerrrArErrZtensorZ sum_reducerr`r r5rWr4r1r r rrrrs<  9  i