code: def encode_and_decode(
      state, bucket
  ) -> torch.futures.Future[torch.Tensor]:
      """
      The Gradient Communication Quantization hook.
      """
      assert (
          state.process_group is not None
      ), "The process_group should be initialized first!"
      process_group = state.process_group
      world_size = dist.get_world_size(process_group)
      hidden_size = state.hidden_size
      quantile_value = state.quantile_value
      quantizer = GCQ(
          world_size=world_size,
          hidden_size=hidden_size,
          bucket=bucket,
          quantile_value=quantile_value,
      )
      scale = quantizer.get_scale()
      # All_reduce scales between ranks and
      # every rank gets the same global scale to encode gradients finally.
      dist.all_reduce(
          scale,
          op=dist.ReduceOp.MAX,
          group=process_group,
          async_op=True,
      ).get_future().wait()
      # Quantize gradients to int8 locally.
      compressed_gradient = quantizer.encode_bucket()
      # All_reduce int8 gradients.
      fut = dist.all_reduce(
          compressed_gradient, group=process_group, async_op=True
      ).get_future()
  
      def decompress(fut):
          """
          Dequantize the all_reduced int8 gradients to original dtype (default: fp16).
          """
          decompressed_gradient = quantizer.decode_bucket()
          dist.barrier(group=process_group, async_op=True)
          return decompressed_gradient
  
      return fut.then(decompress)
  
file_path: lightseq/lightseq/training/gcq/gcq.py
imports: import math
  import torch
  import torch.distributed
label_desc: Use image node_id 1 for calling a global function with example usage: encode_and_decode(state, bucket) and returns: fut, decompressed_gradient