code: def _background_thread_fetch_current_state(
      serializer: SerializerBase,
      pipe: mp.connection.Connection,
      get_current_state_ref: weakref.WeakMethod,
  ):
      """
      Executed in the host process as a background thread. Fetches the averager state when asked by peers.
      :param serializer: a serializer with which to convert metadata into bytes
      :param pipe: DecentralizedAverager's control pipe (from host process side)
      :param get_current_state_ref: a WeakMethod wrapped around DecentralizedAverager.get_current_state (instance-bound)
      """
      while True:
          try:
              trigger, future = pipe.recv()
          except BaseException as e:
              logger.debug(
                  f"Averager background thread finished: {repr(e)}"
              )
              break
  
          if trigger == "_SHUTDOWN":
              break
  
          assert trigger == "_TRIGGER_GET_CURRENT_STATE"
          try:
              get_current_state = get_current_state_ref()
              if get_current_state is None:
                  break
              state = get_current_state()
              assert 0 < len(state) <= 3
              if len(state) != 3:
                  state = tuple(state + (None,) * (3 - len(state)))
              state_metadata, state_tensors, tensor_infos = state
              del get_current_state
  
              state_metadata = serializer.dumps(state_metadata)
              state_tensors = tuple(
                  tensor.cpu()
                  .detach()
                  .requires_grad_(tensor.requires_grad)
                  for tensor in state_tensors
              )
              # note: we cast tensors to CPU on host side to avoid initializing cuda in the guest process
              future.set_result(
                  (state_metadata, state_tensors, tensor_infos)
              )
          except BaseException as e:
              future.set_exception(e)
              logger.warning(e)
              continue
  
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 1 for calling a global function with example usage: _background_thread_fetch_current_state(serializer, pipe, get_current_state_ref) without return types