U Jc@s6ddlmZmZddlZddlmZdddddgZejej ej j ej dd d Z ejej ej j ej d d dZejej ej j ej d d dZejej ej j ej d ddZeeej gej j ej feeej gej j ej fdddZeeej gej j ej feeej gej j ej fdddZdS))AnyCallableNallreduce_hookfp16_compress_hookbf16_compress_hookfp16_compress_wrapperbf16_compress_wrapper) process_grouptensorreturncCs@|dk r |ntjj}||tj||ddddS)zEAverages the input gradient tensor by allreduce and returns a future.NTgroupZasync_opcSs |dSNr)value)futr]/tmp/pip-unpacked-wheel-gikjz4vx/torch/distributed/algorithms/ddp_comm_hooks/default_hooks.pyz _allreduce_fut..)distr WORLDdiv_size all_reduce get_futurethen)r r group_to_userrr_allreduce_futs r)r bucketr cCst||S)a{ This DDP communication hook just calls ``allreduce`` using ``GradBucket`` tensors. Once gradient tensors are aggregated across all workers, its ``then`` callback takes the mean and returns the result. If user registers this hook, DDP results is expected to be same as the case where no hook was registered. Hence, this won't change behavior of DDP and user can use this as a reference or modify this hook to log useful information or any other purposes while unaffecting DDP behavior. Example:: >>> # xdoctest: +SKIP >>> ddp_model.register_comm_hook(process_group, allreduce_hook) )rbuffer)r rrrrrscs\|dk r |ntjj}|}tj|}tj ||dd }fdd}| |S)a/ This DDP communication hook implements a simple gradient compression approach that casts ``GradBucket`` tensor to half-precision floating-point format (``torch.float16``) and then divides it by the process group size. It allreduces those ``float16`` gradient tensors. Once compressed gradient tensors are allreduced, the chained callback ``decompress`` casts it back to the input data type (such as ``float32``). Example:: >>> # xdoctest: +SKIP >>> ddp_model.register_comm_hook(process_group, fp16_compress_hook) NTr cs}||d|SrrZcopy_rrZdecompressed_tensorrrr decompressBsz&fp16_compress_hook..decompress) rr rrrtotorchfloat16rrrrr rrZ world_sizeZcompressed_tensorrr#rr"rr+s  cs\|dk r |ntjj}|}tj|}tj ||dd }fdd}| |S)a Warning: This API is experimental, and it requires NCCL version later than 2.9.6. This DDP communication hook implements a simple gradient compression approach that casts ``GradBucket`` tensor to half-precision `Brain floating point format `_ (``torch.bfloat16``) and then divides it by the process group size. It allreduces those ``bfloat16`` gradient tensors. Once compressed gradient tensors are allreduced, the chained callback ``decompress`` casts it back to the input data type (such as ``float32``). Example:: >>> # xdoctest: +SKIP >>> ddp_model.register_comm_hook(process_group, bf16_compress_hook) NTr cs}||d|Srr r!r"rrr#fsz&bf16_compress_hook..decompress) rr rrrr$r%bfloat16rrrrr'rr"rrLs  )hookr cs$tjtjjtjdfdd }|S)aZ This wrapper casts the input gradient tensor of a given DDP communication hook to half-precision floating point format (``torch.float16``), and casts the resulting tensor of the given hook back to the input data type, such as ``float32``. Therefore, ``fp16_compress_hook`` is equivalent to ``fp16_compress_wrapper(allreduce_hook)``. Example:: >>> # xdoctest: +SKIP >>> state = PowerSGDState(process_group=process_group, matrix_approximation_rank=1, start_powerSGD_iter=10) >>> ddp_model.register_comm_hook(state, fp16_compress_wrapper(powerSGD_hook)) rr cs6tj|}fdd}||S)Ncs}|||SNr r!r"rrr#szMfp16_compress_wrapper..fp16_compress_wrapper_hook..decompress) set_bufferrr$r%r&rZ hook_staterrr#r)r"rfp16_compress_wrapper_hooks  z9fp16_compress_wrapper..fp16_compress_wrapper_hookr GradBucketr%futuresFutureTensor)r)r/rr.rrps cs$tjtjjtjdfdd }|S)a Warning: This API is experimental, and it requires NCCL version later than 2.9.6. This wrapper casts the input gradient tensor of a given DDP communication hook to half-precision `Brain floating point format `_ (``torch.bfloat16``), and casts the resulting tensor of the given hook back to the input data type, such as ``float32``. Therefore, ``bf16_compress_hook`` is equivalent to ``bf16_compress_wrapper(allreduce_hook)``. Example:: >>> # xdoctest: +SKIP >>> state = PowerSGDState(process_group=process_group, matrix_approximation_rank=1, start_powerSGD_iter=10) >>> ddp_model.register_comm_hook(state, bf16_compress_wrapper(powerSGD_hook)) r*cs6tj|}fdd}||S)Ncs}|||Sr+r r!r"rrr#szMbf16_compress_wrapper..bf16_compress_wrapper_hook..decompress)r,rr$r%r(rr-r.r"rbf16_compress_wrapper_hooks  z9bf16_compress_wrapper..bf16_compress_wrapper_hookr0)r)r5rr.rrs )typingrrr%Ztorch.distributedZ distributedr__all__Z ProcessGroupr4r2r3rr1rrrrrrrrrs2    " % %