U Jc @sddlmZddlZddlZddlmZddlZddlZddl m Z ddl m Zddl mZddd gZeeZdd d Zdd d ZddZddZGdddeZee jejjejdddZee jejjejddd ZdS)) defaultdictN)Dict) default_hooks)distributed_c10d PowerSGDState powerSGD_hookbatched_powerSGD_hookc Cst|jdkr"|jd|jdks&t|jd}|jd}|j}|dksX|tjtjfkrft||dn&tjj ||tj ||||j |dfddS) z Decide between Gram-Schmidt or QR factorization to orthogonalize a batch of matrices. QR factorization doesn't work with half-precision, but it is usually faster with a rank > 2. rr)epsilondevicedtypeoutN) lenshapeAssertionErrorrtorchZfloat16Zbfloat16_orthogonalize_gram_schmidtZlinalgZqremptyr)matricesr Z num_matricesZrankrr]/tmp/pip-unpacked-wheel-gikjz4vx/torch/distributed/algorithms/ddp_comm_hooks/powerSGD_hook.py_orthogonalizes&  rc Cs|jd}t|D]}|dddd||df}|dkrz|tj|ddd}Wqtk r|td|dYqXn|tj|ddd|}|d|kr|dddd|ddf}|tj||ddd|8}qdS) z Applies Gram-Schmidt procedure to orthogonalize a batch of matrices. If epsilon is 0, this is equivalent to `torch.qr(matrices, out=(matrices, _))`, r NrrT)ZdimZkeepdimzThe matrices to be orthogonalized has at least a column of all 0s. Please set a small value such as 1e-8 as `orthogonalization_epsilon` in PowerSGD state.g) rrangerZnormZeroDivisionErrorloggererrorfill_sum)rr num_colsicolrestrrrr)s   rcCs&||}|||}|||k||fS)a Returns a recommendation as to whether the 2D tensor described by the arguments is worth compressing, including statistics describing the expected savings from compression. We consider a tensor worth compressing when ``min_compression_rate`` < uncompressed size / compressed size, where uncompressed size = ``num_rows`` * ``num_cols``, and compressed size = (``num_rows`` + ``num_cols``) * ``matrix_approximation_rank``. The result of this function is a tuple of the form (compression_recommendation, uncompressed_el_count, compressed_el_count), where: compresion_recommendation is true if the tensor is worth compressing, and false otherwise (see above); uncompressed_el_count is the uncompressed element count, i.e. ``num_rows`` * ``num_cols``; and, compress_el_count is the element count after compression, i.e. (``num_rows`` + ``num_cols``) * ``matrix_approximation_rank``. r)Znum_rowsr"matrix_approximation_rankmin_compression_rateuncompressed_sizeZcompressed_sizerrr_should_compressIs   r)c CsR|rN|j|jkrN|}td|j|d|d|d|j|j|_dS)zy Report compression stats at the frequency of `compression_stats_logging_frequency` specified in PowerSGD state. z\Compression stats: iter {}, total before compression {}, total after compression {}, rate {}rr rN)is_lastiternext_stats_reportcompression_statsrinfoformat#compression_stats_logging_frequency)bucketstatestatsrrr_report_compression_statsds r4c@sfeZdZdZdddddddd d d d d dddddgZd&edddZddZdd Zd!d"Z d#d$Z d%S)'rah Stores both the algorithm's hyperparameters and the internal state for all the gradients during the training. Particularly, ``matrix_approximation_rank`` and ``start_powerSGD_iter`` are the main hyperparameters that should be tuned by the user. For performance, we suggest to keep binary hyperparameters ``use_error_feedback`` and ``warm_start`` on. 1. ``matrix_approximation_rank`` controls the size of compressed low-rank tensors, which determines the compression rate. The lower the rank, the stronger the compression. 1.1. If ``matrix_approximation_rank`` is too low, the full model quality will need more training steps to reach or will never reach and yield loss in accuracy. 1.2. The increase of ``matrix_approximation_rank`` can substantially increase the computation costs of the compression, and the accuracy may not be futher improved beyond a certain ``matrix_approximation_rank`` threshold. To tune ``matrix_approximation_rank``, we suggest to start from 1 and increase by factors of 2 (like an expoential grid search, 1, 2, 4, ...), until a satisfactory accuracy is reached. Typically only a small value 1-4 is used. For some NLP tasks (as shown in Appendix D of the original paper), this value has been increased to 32. 2. ``start_powerSGD_iter`` defers PowerSGD compression until step ``start_powerSGD_iter``, and vanilla allreduce runs prior to step ``start_powerSGD_iter``. This hybrid scheme of **vanilla allreduce + PowerSGD** can effectively improve the accuracy, even a relatively small ``matrix_approximation_rank`` is used. This is because that, the beginning of training phase is usually very sensitive to inaccurate gradients, and compressing gradients too early may make the training quickly take a suboptimal trajectory, which can result in an irrecoverable impact on the accuracy. To tune ``start_powerSGD_iter``, we suggest to start with 10% of total training steps, and increase it until a satisfactory accuracy is reached. If there is a warm-up stage in the training, ``start_powerSGD_iter`` typically should be no less than the number of warm-up steps. 3. ``min_compression_rate`` is the minimum compression rate required when a layer is compressed. Due to the computation overheads incurred by the compression, a tensor is worth compressing only if there can be sufficient saving in bandwidth, where ``(num_rows + num_cols) * matrix_approximation_rank * min_compression_rate < num_rows * num_cols``. If the specified compression rate threshold cannot be satisfied, the tensor will be directly allreduced without compression. Compression statistics are logged every ``compression_stats_logging_frequency`` iterations once PowerSGD compression starts. 4. ``orthogonalization_epsilon`` can be a very small value (e.g., 1e-8) added to every normalized matrix column in orthogonalization step, to prevent div-by-zero error if any column has all 0s. If this can already be prevented (e.g., by batch normalization), an epsilon of 0 is recommended for accuracy. 5. ``batch_tensors_with_same_shape`` controls whether to compress and decompress tensors with same shape in a batched operation to achieve higher parallelism. Note that you should also increase the bucket size (i.e., ``bucket_cap_mb`` arg in DDP constructor) to make more same-shaped tensors appear in the same bucket, however this may reduce the overlap between computation and communication, and increase the memory footprint due to stacking the tensors of the same shape. Set to ``True`` if the compression / decompression computation is a bottleneck. .. warning :: If error feedback or warm-up is enabled, the minimum value of ``start_powerSGD_iter`` allowed in DDP is 2. This is because there is another internal optimization that rebuilds buckets at iteration 1 in DDP, and this can conflict with any tensor memorized before the rebuild process. process_groupr&start_powerSGD_iterr'orthogonalization_epsilonuse_error_feedback warm_startbatch_tensors_with_same_shaperng error_dict p_memory_dict q_memory_dictr+total_numel_before_compressiontotal_numel_after_compressionr0r,rr Tr'F)r:c Cstd|||||||| | ||_||_|s4|rD|dkrDtd||_||_||_||_ ||_ t j ||_i|_i|_i|_d|_d|_d|_td| |_d|_| |_dS)NaPowerSGD config: matrix_approximation_rank = {}; start_powerSGD_iter = {}; min_compression_rate = {}; orthogonalization_epsilon = {}; use_error_feedback = {}; warm_start = {}; random_seed = {}; compression_stats_logging_frequency = {}; batch_tensors_with_same_shape = {}rzExpect `start_powerSGD_iter` > 1 if `use_error_feedback` or `warm_start` is enabled, because PowerSGD can only be applied after the first two iterations in DDP.r)rr.r/r5r& ValueErrorr6r'r8r9r7nprandomZ RandomStater;r<r=r>r+r?r@maxr0r,r:) selfr5r&r6r'r8r9r7Z random_seedr0r:rrr__init__sJ   zPowerSGDState.__init__cstdfddjDS)z Returns a ``Dict[str, Any]`` which will be pickled and saved. ``process_group`` is not serializable and excluded from a returned state. zHNOTE: Process group is not serializable and excluded from a saved state.cs i|]}|dkr|t|qS)r5)getattr).0slotrGrr sz.PowerSGDState.__getstate__..)rwarning __slots__rLrrLr __getstate__ s  zPowerSGDState.__getstate__cCs6t|_td|D]\}}t|||qdS)zz Takes a provided ``state`` and retrieves ``PowerSGDState``. ``process_group`` is set to default. zNOTE: Process group will be set to a default group (i.e. the world size). If a different group is desired, please set `self.process_group` after PowerSGD state is loaded.N)rZ_get_default_groupr5rrNitemssetattr)rGr2rKvaluerrr __setstate__s  zPowerSGDState.__setstate__cCs8|r|jd7_|j|jkr4td|jdS)Nrz,Start to apply PowerSGD after {} iterations.)r*r+r6rr.r/)rGr1rrrmaybe_increase_iter's   z!PowerSGDState.maybe_increase_itercCs(|jdkr|j|jnd}||j|jfS)a Returns the latest compression statistics as a tuple of the form (compress_rate, numel_before_compression, numel_after_compression), where: compress_rate is the effective compression rate i.e. (number of elements before compression) / (number of elements after compression); numel_before_compression is the total number of elements before compression was applied; and, numel_after_compression is the total number of elements after compression was applied. r)r@r?)rGZ compress_raterrrr-2s zPowerSGDState.compression_statsN) rrAr TTrrrBF) __name__ __module__ __qualname____doc__rOboolrHrPrTrUr-rrrrrtsF   _  )r2r1returnc s6 j}|dk r|ntjj  j jkrL t Sj j } djd} jrĈ jkr jn&td|tj||d j<t}gg d}d}|D]}||jdd} | j\} } t| | j} t| | | j} j| d7_| drr | || | 7}|| | 7} j!| d7_!q | j!| d7_!qt"  rt#dd Dntj$g|d}d } j%r܈ j&kr&d } j%rtd ||tj'||d j&<tj'||d j(<t)t* D]}|j |q2 fd d}g ggd}d}|D]}|j\}} } t| | j} | j&|||| | || |  j(|||| | || | ||| | 7}||| | 7}qr|s:D]}t+| j,q$n^tj-j.gdJt/ j01dD],}|2tj3|jd|dt+| j,q`W5QRXt4 D]\}}}tj5|||dqtj6|d d7} fdd} fdd} f dd}|8|8|8|S)aL This DDP communication hook implements PowerSGD gradient compression algorithm described in the `paper `_. Once gradient tensors are aggregated across all workers, this hook applies compression as follows: 1. Views the input flattened 1D gradient tensor as a list of per-parameter tensors, and divides all the tensors into two groups: 1.1 The tensors that should be compressed before allreduce, because the compression can give enough saving in bandwidth. 1.2 Rest of the tensors will be directly allreduced without compression, including all the vector tensors (for biases). 2. Handles uncompressed tensors: 2.1. Allocate contiguous memory for those uncompressed tensors, and allreduces all the uncompressed tensors as a batch, without compression; 2.2. Copies the individual uncompressed tensors from the contiguous memory back to the input tensor. 3. Handles the tensors that should be compressed by PowerSGD compression: 3.1. For each tensor M, creates two low-rank tensors P and Q for decomposing M, such that M = PQ^T, where Q is initialized from a standard normal distribution and orthogonalized; 3.2. Computes each P in Ps, which is equal to MQ; 3.3. Allreduces Ps as a batch; 3.4. Orthogonalizes each P in Ps; 3.5. Computes each Q in Qs, which is approximately equal to M^TP; 3.6. Allreduces Qs as a batch; 3.7. Computes each M among all the compressed tensors, which is approximately equal to PQ^T. Note that this communication hook enforces vanilla allreduce for the first ``state.start_powerSGD_iter`` iterations. This not only gives the user more control over the tradeoff between speedup and accuracy, but also helps abstract away some complexity of the internal optimization of DDP for future communication hook developers. Args: state (PowerSGDState): State information to configure the compression rate and support error feedback, warm start, etc. To tune the compression configs, mainly need to tune ``matrix_approximation_rank``, ``start_powerSGD_iter`` and ``min_compression_rate``. bucket (dist.GradBucket): Bucket that stores a 1D flattened gradient tensor that batches multiple per-variable tensors. Note that since DDP comm hook only supports single process single device mode, only exactly one tensor is stored in this bucket. Returns: Future handler of the communication, which updates the gradients in place. Example:: >>> # xdoctest: +SKIP >>> state = PowerSGDState(process_group=process_group, matrix_approximation_rank=1, start_powerSGD_iter=10, min_compression_rate=0.5) >>> ddp_model.register_comm_hook(state, powerSGD_hook) NrBA zero tensor of length {} that represents local error is created.r rr cSsg|]}|dqS)r])view)rJtensorrrr sz!powerSGD_hook..FTzXAllocating contiguous memory of length {} for Ps, and of length {} for Qs, respectively.c3s^D]P}jrBt|}|dkr4|ddVqXt|Vq|D]}|dVqFqdS)Nrr)valuesr:rZ unsqueezerstack)tensors batch_sizer_)shape_to_tensorsr2rr!maybe_batched_tensors_to_compresss z8powerSGD_hook..maybe_batched_tensors_to_compressZdevicesʚ;cpurgroupZasync_opcsn|d}d}D]0}||||||||7}qtjjdd dS)NrTrj) rSdiv_copy_ZnumelZview_asdist all_reducer= get_futurewait)futuncompressed_tensors_memoryidxr_) bucket_index group_to_user2uncompressed_tensors world_sizerr,unpack_uncompressed_tensors_and_allreduce_ps/s zCpowerSGD_hook..unpack_uncompressed_tensors_and_allreduce_pscsv|j<D]}t|jqtD]"\}}}tj|dd||dq0tj j dd dS)Nrr rTrjr) rSr=rr7ziprbmm transposernror>rprq)rrpr_q)rurvpsqsr2tensors_to_compressrr compute_qsAs z!powerSGD_hook..compute_qscs| j<t D]"\}}}tj||dd|dq jr D]F}|jddkrbqN|jdd}t |D]\}}| ||q|qNtj rtj jrj<jsڈjjS)Nrr rr)rSrlr>rzrr{r|r:r enumeratermcuda is_available synchronizer8r<r9r=clearrU)rrr}r~r_Zoriginal_tensorsr#Zoriginal_tensor) r1rur input_tensorinput_tensor_cprrrer2rrxrr decompressWs&     z!powerSGD_hook..decompress)9r5rnrkWORLDsizebufferr+r6rUdefault_allreduce_futrrindexrr8r<add_rr.r/rzerosclonedetachZ gradientsr^minr&r)r'r?appendr@r4catr_r9r=rr>rlistrr7rEfork_rng manual_seedr;randintrmrandnrzr{rorpthen)r2r1r5r total_lengthrcZ total_Ps_sizeZ total_Qs_sizer_matrixnmr&Z compress_testrsZneed_randomize_qsrfZp_idxZq_idxrdr~r}Z-allreduce_contiguous_uncompressed_tensors_futryrrr) r1rurrvrrrrrer2rrwrxrrHs;                      c sj}|dk r|ntjj jjkrLt Sj j d j 7_ tt jjd7_d}| |dtdjr&jkrjn(td|tj|jdj<t !"j#rFj$krj#rbtdjfdd}|d j%d j$<|d j%d j&<t'j&tj(j&j$d tj)j$d d *}fdd} f dd}|+|+|S)a This DDP communication hook implements a simplified PowerSGD gradient compression algorithm described in the `paper `_. This variant does not compress the gradients layer by layer, but instead compresses the flattened input tensor that batches all the gradients. Therefore, it is **faster** than :meth:`powerSGD_hook`, but usually results in a **much lower accuracy**, unless ``matrix_approximation_rank`` is 1. .. warning :: Increasing ``matrix_approximation_rank`` here may not necessarily increase the accuracy, because batching per-parameter tensors without column/row alignment can destroy low-rank structure. Therefore, the user should always consider :meth:`powerSGD_hook` first, and only consider this variant when a satisfactory accuracy can be achieved when ``matrix_approximation_rank`` is 1. Once gradient tensors are aggregated across all workers, this hook applies compression as follows: 1. Views the input flattened 1D gradient tensor as a square-shaped tensor M with 0 paddings; 2. Creates two low-rank tensors P and Q for decomposing M, such that M = PQ^T, where Q is initialized from a standard normal distribution and orthogonalized; 3. Computes P, which is equal to MQ; 4. Allreduces P; 5. Orthogonalizes P; 6. Computes Q, which is approximately equal to M^TP; 7. Allreduces Q; 8. Computes M, which is approximately equal to PQ^T. 9. Truncates the input tensor to the original length. Note that this communication hook enforces vanilla allreduce for the first ``state.start_powerSGD_iter`` iterations. This not only gives the user more control over the tradeoff between speedup and accuracy, but also helps abstract away some complexity of the internal optimization of DDP for future communication hook developers. Args: state (PowerSGDState): State information to configure the compression rate and support error feedback, warm start, etc. To tune the compression configs, mainly need to tune ``matrix_approximation_rank`` and ``start_powerSGD_iter``. bucket (dist.GradBucket): Bucket that stores a 1D flattened gradient tensor that batches multiple per-variable tensors. Note that since DDP comm hook only supports single process single device mode, only exactly one tensor is stored in this bucket. Returns: Future handler of the communication, which updates the gradients in place. Example:: >>> # xdoctest: +SKIP >>> state = PowerSGDState(process_group=process_group, matrix_approximation_rank=1) >>> ddp_model.register_comm_hook(state, batched_powerSGD_hook) Nrr r\r zLInitializing low-rank tensors P and Q, each of which has a shape of {} x {}.c sn|rTtjjgd:t|dtjjdjdW5QRSQRXntj jjdSdS)zOReturns a low-rank 2D tensor of square_side_length * matrix_approximation_rank.rgrhrir N) rrErrrrr&rtorZfill_random_valuesr;)rrsquare_side_lengthr2rrcreate_low_rank_tensors"z5batched_powerSGD_hook..create_low_rank_tensorFrTrrjcsb|dj<tjtjjjdtjjdd dS)NrrTrj) rSr=rrmatmultr>rnrorprq)rr)rurvrr2rr compute_qs  z(batched_powerSGD_hook..compute_qcs|j<tjjjdjrHj<tj r^tj j sxj j }|S)Nr)rSrlr>rrr=rr8r<rrrr9rresize_rU)rrret) r1rurrrrr2rrxrrr3s        z)batched_powerSGD_hook..decompress),r5rnrkrrrr+r6rUrrrrr?mathceilsqrtr@r&rr r4rr8r<rrr.r/rrrrrr^r9r=r;r>rrrorpr)r2r1r5Zpadded_total_lengthrZallreduce_p_futrrr) r1rurrvrrrrr2rrxrr ~s9             )r)r) collectionsrloggingrtypingrZnumpyrDrZtorch.distributedZ distributedrnrrr__all__ getLoggerrVrrrr)r4objectrZ GradBucketZfuturesZFutureZTensorrr rrrrs:        V 9