class_docstr: An internal class that runs butterfly AllReduce in a predefined group of averagers.
  
  This class inherits hivemind.p2p.ServicerBase, so it can be used as an RPCServicer for testing purposes without
  creating a full DecentralizedAverager.
  
  :note: this class returns **differences** between averaged and local tensors in order to improve numerical stability
  :param p2p: a hivemind.p2p.P2P instance used for communication with other peers
  :param servicer_type: a hivemind.p2p.ServicerBase subclass whose RPC signatures are used
    when requesting other peers. Typically, it is DecentralizedAverager, its derivative,
    or AllReduceRunner itself (for testing purposes).
  :param prefix: namespace for servicer's RPCs (typically, equal to prefix for group keys)
  :param group_id: unique identifier of this specific all-reduce run
  :param tensors: local tensors that should be averaged with groupmates
  :param weight: scalar weight of this peer's tensors in the average (doesn't need to sum up to 1)
  :param peer_id: your peer_id, must be included in ordered_peer_ids
  :param ordered_peer_ids: group peer_ids ordered s.t. i-th peer_id is responsible for averaging i-th part
  :param peer_fractions: for each peer, a target fraction of vector elements that this peer should average
    (the actual number of values by peer will be nearly proportional, but there are no exact guarantees)
  :param modes: AveragingMode for each peer in ordered_peer_ids (normal, client-only or auxiliary)
  :param sender_timeout: during all_reduce, any sender that fails to send tensor chunk within this many seconds from
    previous chunk will be marked as failed and excluded from averaging. default: equal to next_chunk_timeout
  :param reducer_timeout: during all_reduce, any reducer that fails to send results chunk within this many seconds
    from previous chunk will be marked as failed and excluded from averaging. default: 2 x sender_timeout
  :param kwargs: additional parameters (e.g. part_size_bytes) will be passed to TensorPartContainer
  :note: Full-mode peers send and receive tensor parts concurrently, assuming a full-duplex TCP stream. In turn,
    non-averaging peers receive results only after they finish sending, which helps them avoid
    throughput issues in case of asymmetric high-latency connections (e.g. ACK compression).
code: def __init__(
      self,
      *,
      p2p: P2P,
      servicer_type: Type[ServicerBase],
      prefix: Optional[str],
      group_id: GroupID,
      tensors: Sequence[torch.Tensor],
      weight: Optional[float] = None,
      ordered_peer_ids: Sequence[PeerID],
      peer_fractions: Tuple[float, ...],
      modes: Optional[Sequence[AveragingMode]] = None,
      sender_timeout: Optional[float] = None,
      reducer_timeout: Optional[float] = None,
      **kwargs,
  ):
      self._p2p = p2p
      self.peer_id = p2p.peer_id
      assert (
          self.peer_id in ordered_peer_ids
      ), "peer_id is not a part of the group"
      if reducer_timeout is not None and (
          sender_timeout is None or reducer_timeout <= sender_timeout
      ):
          raise ValueError(
              "If reducer_timeout is enabled, sender_timeout must be shorter than reducer_timeout. "
              "Otherwise, there is a chance that reducers will be banned while they await senders."
          )
  
      if not issubclass(servicer_type, ServicerBase):
          raise TypeError(
              "`servicer_type` is expected to be a ServicerBase subclass"
          )
      self._servicer_type = servicer_type
      self._prefix = prefix
  
      modes = modes or tuple(
          AveragingMode.CLIENT if frac == 0 else AveragingMode.NODE
          for frac in peer_fractions
      )
      assert len(modes) == len(
          ordered_peer_ids
      ), "lists have inconsistent length"
      assert any(
          mode != AveragingMode.CLIENT for mode in modes
      ), "cannot run allreduce without reducers"
      for mode, frac in zip(modes, peer_fractions):
          assert (
              mode != AveragingMode.CLIENT or frac == 0
          ), "client-mode peer should have zero all-reduce fraction"
  
      self.group_id, self.ordered_peer_ids = group_id, ordered_peer_ids
      self.modes, self.peer_fractions = modes, peer_fractions
  
      if weight is None:
          weight = float(
              modes[self.ordered_peer_ids.index(self.peer_id)]
              != AveragingMode.AUX
          )
      self.weight = weight
  
      self._future = asyncio.Future()
  
      self.sender_peer_ids = []
      for peer_id, mode in zip(self.ordered_peer_ids, modes):
          if mode != AveragingMode.AUX:
              self.sender_peer_ids.append(peer_id)
  
      self.sender_timeout, self.reducer_timeout = (
          sender_timeout,
          reducer_timeout,
      )
      self.all_senders_started = asyncio.Event()
      self.banned_senders: Set[
          PeerID
      ] = set()  # peers that did not send data by next_chunk_timeout
      self.banlock = asyncio.Lock()
  
      self.active_senders: Set[
          PeerID
      ] = set()  # peers that began sending data via rpc_aggregate_part
      if self.peer_id in self.sender_peer_ids:
          self.active_senders.add(self.peer_id)
      if len(self.active_senders) == len(self.sender_peer_ids):
          self.all_senders_started.set()
  
      peer_id_index = self.ordered_peer_ids.index(self.peer_id)
      self.tensor_part_container = TensorPartContainer(
          tensors, peer_fractions, return_deltas=True, **kwargs
      )
      self.parts_for_local_averaging = (
          self.tensor_part_container.get_raw_input_parts(peer_id_index)
      )
      self.tensor_part_reducer = TensorPartReducer(
          tuple(part.shape for part in self.parts_for_local_averaging),
          len(self.sender_peer_ids),
      )
  
imports: import asyncio
  from enum import Enum
  from typing import AsyncIterator, Optional, Sequence, Set, Tuple, Type
  import torch
  from hivemind.averaging.partition import AllreduceException, BannedException, TensorPartContainer, TensorPartReducer
  from hivemind.compression import deserialize_torch_tensor, serialize_torch_tensor
  from hivemind.p2p import P2P, P2PContext, PeerID, ServicerBase, StubBase
  from hivemind.proto import averaging_pb2
  from hivemind.utils import get_logger
  from hivemind.utils.asyncio import achain, aiter_with_timeout, amap_in_executor, anext, as_aiter, attach_event_on_finished
label_desc: Use image node_id 1 to create a new AllReduceRunner object from inherited base classes: ServicerBase with example: obj = AllReduceRunner()