code: def __init__(
      self,
      params,
      named_parameters,
      compression,
      backward_passes_per_step=1,
  ):
      super(self.__class__, self).__init__(params)
      self._compression = compression
  
      if named_parameters is not None:
          named_parameters = list(named_parameters)
      else:
          named_parameters = []
  
      self._enable_async = int(os.getenv("BYTEPS_ENABLE_ASYNC", 0)) != 0
      if self._enable_async:
          assert (
              int(os.getenv("DMLC_NUM_WORKER")) > 1
          ), "Async is only valid for distributed training"
          print("BytePS: enable asynchronous training")
  
      # make sure that named_parameters are tuples
      if any([not isinstance(p, tuple) for p in named_parameters]):
          raise ValueError(
              "named_parameters should be a sequence of "
              "tuples (name, parameter), usually produced by "
              "model.named_parameters()."
          )
  
      dups = _DistributedOptimizer.find_duplicates(
          [k for k, _ in named_parameters]
      )
      if len(dups) > 0:
          raise ValueError(
              "Parameter names in named_parameters must be unique. "
              "Found duplicates: %s" % ", ".join(dups)
          )
  
      if len(named_parameters) > 0:
          if isinstance(named_parameters[0][1], torch.Tensor):
              if any(
                  [
                      not isinstance(p, torch.Tensor)
                      for name, p in named_parameters
                  ]
              ):
                  raise ValueError(
                      "named_parameters should consistently be a sequence of "
                      "tuples (name, torch.Tensor)"
                  )
              self._is_tensor_instance = True
              # there is an issue when using torch.Tensor as key, so use its hash instead
              # https://github.com/pytorch/pytorch/issues/7733
              self._parameter_names = {
                  v.__hash__(): k for k, v in sorted(named_parameters)
              }
              self._tensor_list = [
                  tensor for name, tensor in named_parameters
              ]
          else:
              self._is_tensor_instance = False
              self._parameter_names = {
                  v: k for k, v in sorted(named_parameters)
              }
      else:
          self._is_tensor_instance = False
          self._parameter_names = {
              v: "push_pull.noname.%s" % i
              for param_group in self.param_groups
              for i, v in enumerate(param_group["params"])
          }
      self.backward_passes_per_step = backward_passes_per_step
      self._push_pull_delay = {
          v: self.backward_passes_per_step
          for _, v in sorted(named_parameters)
      }
      self._handles = {}
      self._grad_accs = []
      self._requires_update = set()
      self._should_sync = True
      if size() > 1:
          self._register_hooks()
  
      # declare tensors
      for name in sorted(self._parameter_names.values()):
          declare("Gradient." + name)
      # We use two loops for load-balancing
      for name in sorted(self._parameter_names.values()):
          declare("Parameter." + name)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from contextlib import contextmanager
  from byteps.torch.compression import Compression
  from byteps.torch.ops import push_pull_async_inplace
  from byteps.torch.ops import push_pull
  from byteps.torch.ops import poll, synchronize, declare
  from byteps.torch.ops import init, shutdown, suspend, resume
  from byteps.torch.ops import size, local_size, rank, local_rank
  import os
  import torch
  import collections
  import io
  import cloudpickle
label_desc: Use image node_id 1 to create a new _DistributedOptimizer object from inherited base classes: torch.optim with example: obj = _DistributedOptimizer(params, named_parameters, compression, backward_passes_per_step)