class_docstr: Parameter averaging service. A trainer can run this service in background to periodically average his parameters
  with other trainers. The averaging pattern is chosen so that (1) you only need to average with a small
  group of peers at a time, but (2) all trainers will converge to global average in a logarithmic number of steps.
  
  :param averaged_tensors: a sequence of pytorch tensors that will be averaged in each all-reduce
  :param dht: a DHT node that will be used to find groups
  :param start: if True, starts the background process immediately
  
  :param prefix: a shared prefix for all group keys
  :param target_group_size: attempts to form groups with up to this many peers (recommended: a power of 2, e.g. 16)
  :param initial_group_bits: a string of bits ('0' and '1') that define the initial group key (bucket index)
  :param min_matchmaking_time: when looking for group, wait for requests for at least this many seconds
  :param compression: optionally compress tensors with this compression algorithm before running all-reduce
  :param state_compression: a separate compression strategy for load_state_from_peers (default = no compression)
  :param tensor_infos: CompressionInfo for each respective tensor; this determines how the tensor will be compressed
  :param averaging_alpha: optional "learning rate" for averaging. If specified, local parameters will be shifted
    towards the (estimated) average by this coefficient. By default, local parameters are set equal to average.
  :param request_timeout: when looking for group, wait for a response from leader for at most this many seconds.
  :note: request_timeout must be smaller than min_matchmaking_time to avoid potential deadlocks.
  :param part_size_bytes: tensors for AllReduce are processed in parts of up to this size (after compression)
  :param bandwidth: if specified, this value represents the network bandwidth available to averager.
        By default, the averager is assumed to have the average bandwidth of his group.
        If bandwidth == 0, averager will rely on its groupmates to do all the averaging.
  :param client_mode: if False, this averager will accept incoming requests from other peers.
        if True, the averager will only join existing groups where at least one peer has client_mode=False.
        By default, this flag is copied from DHTNode inside the ``dht`` instance.
  :param auxiliary: if this flag is specified, averager.step will only assist others without sending
        local tensors for averaging
  :param allow_state_sharing: if set to True, other peers can download this peer's state. Can be overwritten
    with averager.allow_state_sharing = True / False
  :param declare_state_period: re-declare averager as a donor for load_state_from_peers every this many seconds
  :param allreduce_timeout: spend at most this many seconds for allreduce (after group is formed)
  :param next_chunk_timeout: during all-reduce and load_state_from_peers, if peer does not send next data chunk in
    this number of seconds, consider it failed and proceed with remaining peers. default: no timeout
  :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 * sender_timeout
  :param shutdown_timeout: when calling .shutdown, wait for up to this many seconds before terminating
  
  Example:
  
  >>> averager = DecentralizedAverager(...)
  >>> with averager.get_tensors() as tensors:
  >>>     # run some code, modify tensors if necessary
  >>>     tensors[0] += 1
  >>> # do not use tensors after the lock is released
  >>> metadata = averager.step(gather=dict(my_batch_size=32))
  >>> # run averaging once (in-place), gather metadata from groupmates
  >>> with averager.get_tensors() as tensors_after_averaging:
  >>>     pass # use the averaged tensors
class_docstr_tok: ["Parameter","averaging","service",".","A","trainer","can","run","this","service","in","background","to","periodically","average","his","parameters","with","other","trainers",".","The","averaging","pattern","is","chosen","so","that","(","1",")","you","only","need","to","average","with","a","small","group","of","peers","at","a","time",",","but","(","2",")","all","trainers","will","converge","to","global","average","in","a","logarithmic","number","of","steps",".",":","param","averaged_tensors",":","a","sequence","of","pytorch","tensors","that","will","be","averaged","in","each","all-reduce",":","param","dht",":","a","DHT","node","that","will","be","used","to","find","groups",":","param","start",":","if","True",",","starts","the","background","process","immediately",":","param","prefix",":","a","shared","prefix","for","all","group","keys",":","param","target_group_size",":","attempts","to","form","groups","with","up","to","this","many","peers","(","recommended",":","a","power","of","2",",","e.g",".","16",")",":","param","initial_group_bits",":","a","string","of","bits","(","'","0","'","and","'","1","'",")","that","define","the","initial","group","key","(","bucket","index",")",":","param","min_matchmaking_time",":","when","looking","for","group",",","wait","for","requests","for","at","least","this","many","seconds",":","param","compression",":","optionally","compress","tensors","with","this","compression","algorithm","before","running","all-reduce",":","param","state_compression",":","a","separate","compression","strategy","for","load_state_from_peers","(","default","=","no","compression",")",":","param","tensor_infos",":","CompressionInfo","for","each","respective","tensor",";","this","determines","how","the","tensor","will","be","compressed",":","param","averaging_alpha",":","optional","``","learning","rate","''","for","averaging",".","If","specified",",","local","parameters","will","be","shifted","towards","the","(","estimated",")","average","by","this","coefficient",".","By","default",",","local","parameters","are","set","equal","to","average",".",":","param","request_timeout",":","when","looking","for","group",",","wait","for","a","response","from","leader","for","at","most","this","many","seconds",".",":","note",":","request_timeout","must","be","smaller","than","min_matchmaking_time","to","avoid","potential","deadlocks",".",":","param","part_size_bytes",":","tensors","for","AllReduce","are","processed","in","parts","of","up","to","this","size","(","after","compression",")",":","param","bandwidth",":","if","specified",",","this","value","represents","the","network","bandwidth","available","to","averager",".","By","default",",","the","averager","is","assumed","to","have","the","average","bandwidth","of","his","group",".","If","bandwidth","==","0",",","averager","will","rely","on","its","groupmates","to","do","all","the","averaging",".",":","param","client_mode",":","if","False",",","this","averager","will","accept","incoming","requests","from","other","peers",".","if","True",",","the","averager","will","only","join","existing","groups","where","at","least","one","peer","has","client_mode=False",".","By","default",",","this","flag","is","copied","from","DHTNode","inside","the","``","dht","``","instance",".",":","param","auxiliary",":","if","this","flag","is","specified",",","averager.step","will","only","assist","others","without","sending","local","tensors","for","averaging",":","param","allow_state_sharing",":","if","set","to","True",",","other","peers","can","download","this","peer","'s","state",".","Can","be","overwritten","with","averager.allow_state_sharing","=","True","\/","False",":","param","declare_state_period",":","re-declare","averager","as","a","donor","for","load_state_from_peers","every","this","many","seconds",":","param","allreduce_timeout",":","spend","at","most","this","many","seconds","for","allreduce","(","after","group","is","formed",")",":","param","next_chunk_timeout",":","during","all-reduce","and","load_state_from_peers",",","if","peer","does","not","send","next","data","chunk","in","this","number","of","seconds",",","consider","it","failed","and","proceed","with","remaining","peers",".","default",":","no","timeout",":","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","*","sender_timeout",":","param","shutdown_timeout",":","when","calling",".shutdown",",","wait","for","up","to","this","many","seconds","before","terminating","Example",":",">",">",">","averager","=","DecentralizedAverager","(","...",")",">",">",">","with","averager.get_tensors","(",")","as","tensors",":",">",">",">","#","run","some","code",",","modify","tensors","if","necessary",">",">",">","tensors","[","0","]","+=","1",">",">",">","#","do","not","use","tensors","after","the","lock","is","released",">",">",">","metadata","=","averager.step","(","gather=dict","(","my_batch_size=32",")",")",">",">",">","#","run","averaging","once","(","in-place",")",",","gather","metadata","from","groupmates",">",">",">","with","averager.get_tensors","(",")","as","tensors_after_averaging",":",">",">",">","pass","#","use","the","averaged","tensors"]
docstr: Run averager function in a background thread; this is needed to avoid a heisenbug with broken OMP on fork
  Turns out, using a non-main thread creates a separate OMP pool that works even if the original pool is corrupted
  Read more: https://github.com/pytorch/pytorch/issues/17199
code: def run(self):
      """
      Run averager function in a background thread; this is needed to avoid a heisenbug with broken OMP on fork
      Turns out, using a non-main thread creates a separate OMP pool that works even if the original pool is corrupted
      Read more: https://github.com/pytorch/pytorch/issues/17199
      """
      thread = threading.Thread(target=self._run_internal, daemon=True)
      thread.start()
      thread.join()
  
imports: from __future__ import annotations
  import asyncio
  import contextlib
  import ctypes
  import multiprocessing
  import os
  import random
  import signal
  import threading
  import weakref
  from dataclasses import asdict
  from typing import Any, AsyncIterator, Dict, Optional, Sequence, Tuple, Union
  import numpy
  import torch
  from hivemind.averaging.allreduce import AllreduceException, AllReduceRunner, AveragingMode, GroupID
  from hivemind.averaging.control import AveragingStage, StepControl
  from hivemind.averaging.group_info import GroupInfo
  from hivemind.averaging.load_balancing import load_balance_peers
  from hivemind.averaging.matchmaking import Matchmaking, MatchmakingException
  from hivemind.averaging.partition import DEFAULT_PART_SIZE_BYTES
  from hivemind.compression import CompressionBase, CompressionInfo, NoCompression, deserialize_torch_tensor
  from hivemind.dht import DHT, DHTID
  from hivemind.p2p import P2P, P2PContext, P2PDaemonError, P2PHandlerError, PeerID, ServicerBase
  from hivemind.proto import averaging_pb2
  from hivemind.utils import MPFuture, TensorDescriptor, get_logger
  from hivemind.utils.asyncio import achain, aiter_with_timeout, anext, as_aiter, azip, enter_asynchronously, switch_to_uvloop
  from hivemind.utils.serializer import MSGPackSerializer, SerializerBase
  from hivemind.utils.streaming import combine_from_streaming, split_for_streaming
  from hivemind.utils.timed_storage import DHTExpiration, ValueWithExpiration, get_dht_time
label_desc: Use image node_id 8 for calling the DecentralizedAverager obj's underlying member method code with example usage: obj.run() without return types