class_docstr: Server allows you to host "experts" - pytorch subnetworks that can be accessed remotely by peers.
  After creation, a server should be started: see Server.run or Server.run_in_background.
  
  A working server does two things:
   - processes incoming forward/backward requests via Runtime (created by the server)
   - publishes updates to expert status every :update_period: seconds
  
  :type dht: an instance of hivemind.DHT. Server will use DHT for all network interactions.
  :param module_backends: dict{expert uid (str) : ModuleBackend} for all expert hosted by this server.
  :param num_connection_handlers: maximum number of simultaneous requests. Please note that the default value of 1
      if too small for normal functioning, we recommend 4 handlers per expert backend.
  :param update_period: how often will server attempt to publish its state (i.e. experts) to the DHT;
      if dht is None, this parameter is ignored.
  :param expiration: when server declares its experts to the DHT, these entries will expire after this many seconds
  :param start: if True, the server will immediately start as a background thread and returns control after server
      is ready (see .ready below)
class_docstr_tok: ["Server","allows","you","to","host","``","experts","''","-","pytorch","subnetworks","that","can","be","accessed","remotely","by","peers",".","After","creation",",","a","server","should","be","started",":","see","Server.run","or","Server.run_in_background",".","A","working","server","does","two","things",":","-","processes","incoming","forward\/backward","requests","via","Runtime","(","created","by","the","server",")","-","publishes","updates","to","expert","status","every",":","update_period",":","seconds",":","type","dht",":","an","instance","of","hivemind.DHT",".","Server","will","use","DHT","for","all","network","interactions",".",":","param","module_backends",":","dict","{","expert","uid","(","str",")",":","ModuleBackend","}","for","all","expert","hosted","by","this","server",".",":","param","num_connection_handlers",":","maximum","number","of","simultaneous","requests",".","Please","note","that","the","default","value","of","1","if","too","small","for","normal","functioning",",","we","recommend","4","handlers","per","expert","backend",".",":","param","update_period",":","how","often","will","server","attempt","to","publish","its","state","(","i.e",".","experts",")","to","the","DHT",";","if","dht","is","None",",","this","parameter","is","ignored",".",":","param","expiration",":","when","server","declares","its","experts","to","the","DHT",",","these","entries","will","expire","after","this","many","seconds",":","param","start",":","if","True",",","the","server","will","immediately","start","as","a","background","thread","and","returns","control","after","server","is","ready","(","see",".ready","below",")"]
docstr: Instantiate a server with several identical modules. See argparse comments below for details
  
  :param num_experts: run this many identical experts
  :param expert_pattern: a string pattern or a list of expert uids,  example: myprefix.[0:32].[0:256]           means "sample random experts between myprefix.0.0 and myprefix.255.255;
  :param expert_uids: spawn experts with these exact uids, overrides num_experts and expert_pattern
  :param expert_cls: expert type from hivemind.moe.server.layers, e.g. 'ffn' or 'transformer';
  :param hidden_dim: main dimension for expert_cls
  :param num_handlers: server will use this many parallel processes to handle incoming requests
  :param min_batch_size: total num examples in the same batch will be greater than this value
  :param max_batch_size: total num examples in the same batch will not exceed this value
  :param device: all experts will use this device in torch notation; default: cuda if available else cpu
  
  :param optim_cls: uses this optimizer to train all experts
  :param scheduler: if not `none`, the name of the expert LR scheduler
  :param num_warmup_steps: the number of warmup steps for LR schedule
  :param num_training_steps: the total number of steps for LR schedule
  :param clip_grad_norm: maximum gradient norm used for clipping
  
  :param initial_peers: multiaddrs of one or more active DHT peers (if you want to join an existing DHT)
  
  :param checkpoint_dir: directory to save and load expert checkpoints
  
  :param compression: if specified, use this compression to pack all inputs, outputs and gradients by all experts
      hosted on this server. For a more fine-grained compression, start server in python and specify compression
      for each BatchTensorProto in ModuleBackend for the respective experts.
  
  :param start: if True, starts server right away and returns when server is ready for requests
  :param stats_report_interval: interval between two reports of batch processing performance statistics
  :param kwargs: any other params will be forwarded to DHT upon creation
docstr_tok: ["Instantiate","a","server","with","several","identical","modules",".","See","argparse","comments","below","for","details",":","param","num_experts",":","run","this","many","identical","experts",":","param","expert_pattern",":","a","string","pattern","or","a","list","of","expert","uids",",","example",":","myprefix.","[","0:32","]",".","[","0:256","]","means","``","sample","random","experts","between","myprefix.0.0","and","myprefix.255.255",";",":","param","expert_uids",":","spawn","experts","with","these","exact","uids",",","overrides","num_experts","and","expert_pattern",":","param","expert_cls",":","expert","type","from","hivemind.moe.server.layers",",","e.g",".","'ffn","'","or","'transformer","'",";",":","param","hidden_dim",":","main","dimension","for","expert_cls",":","param","num_handlers",":","server","will","use","this","many","parallel","processes","to","handle","incoming","requests",":","param","min_batch_size",":","total","num","examples","in","the","same","batch","will","be","greater","than","this","value",":","param","max_batch_size",":","total","num","examples","in","the","same","batch","will","not","exceed","this","value",":","param","device",":","all","experts","will","use","this","device","in","torch","notation",";","default",":","cuda","if","available","else","cpu",":","param","optim_cls",":","uses","this","optimizer","to","train","all","experts",":","param","scheduler",":","if","not","`","none","`",",","the","name","of","the","expert","LR","scheduler",":","param","num_warmup_steps",":","the","number","of","warmup","steps","for","LR","schedule",":","param","num_training_steps",":","the","total","number","of","steps","for","LR","schedule",":","param","clip_grad_norm",":","maximum","gradient","norm","used","for","clipping",":","param","initial_peers",":","multiaddrs","of","one","or","more","active","DHT","peers","(","if","you","want","to","join","an","existing","DHT",")",":","param","checkpoint_dir",":","directory","to","save","and","load","expert","checkpoints",":","param","compression",":","if","specified",",","use","this","compression","to","pack","all","inputs",",","outputs","and","gradients","by","all","experts","hosted","on","this","server",".","For","a","more","fine-grained","compression",",","start","server","in","python","and","specify","compression","for","each","BatchTensorProto","in","ModuleBackend","for","the","respective","experts",".",":","param","start",":","if","True",",","starts","server","right","away","and","returns","when","server","is","ready","for","requests",":","param","stats_report_interval",":","interval","between","two","reports","of","batch","processing","performance","statistics",":","param","kwargs",":","any","other","params","will","be","forwarded","to","DHT","upon","creation"]
code: def create(
      cls,
      num_experts: int = None,
      expert_uids: str = None,
      expert_pattern: str = None,
      expert_cls="ffn",
      hidden_dim=1024,
      optim_cls=torch.optim.Adam,
      scheduler: str = "none",
      num_warmup_steps=None,
      num_training_steps=None,
      clip_grad_norm=None,
      num_handlers=None,
      min_batch_size=1,
      max_batch_size=4096,
      device=None,
      initial_peers=(),
      checkpoint_dir: Optional[Path] = None,
      compression=CompressionType.NONE,
      stats_report_interval: Optional[int] = None,
      custom_module_path=None,
      update_period: float = 30,
      expiration: Optional[float] = None,
      *,
      start: bool,
      **kwargs,
  ) -> Server:
      """
          Instantiate a server with several identical modules. See argparse comments below for details
  
          :param num_experts: run this many identical experts
          :param expert_pattern: a string pattern or a list of expert uids,  example: myprefix.[0:32].[0:256]\
             means "sample random experts between myprefix.0.0 and myprefix.255.255;
          :param expert_uids: spawn experts with these exact uids, overrides num_experts and expert_pattern
          :param expert_cls: expert type from hivemind.moe.server.layers, e.g. 'ffn' or 'transformer';
          :param hidden_dim: main dimension for expert_cls
          :param num_handlers: server will use this many parallel processes to handle incoming requests
          :param min_batch_size: total num examples in the same batch will be greater than this value
          :param max_batch_size: total num examples in the same batch will not exceed this value
          :param device: all experts will use this device in torch notation; default: cuda if available else cpu
  
          :param optim_cls: uses this optimizer to train all experts
          :param scheduler: if not `none`, the name of the expert LR scheduler
          :param num_warmup_steps: the number of warmup steps for LR schedule
          :param num_training_steps: the total number of steps for LR schedule
          :param clip_grad_norm: maximum gradient norm used for clipping
  
          :param initial_peers: multiaddrs of one or more active DHT peers (if you want to join an existing DHT)
  
          :param checkpoint_dir: directory to save and load expert checkpoints
  
          :param compression: if specified, use this compression to pack all inputs, outputs and gradients by all experts
              hosted on this server. For a more fine-grained compression, start server in python and specify compression
              for each BatchTensorProto in ModuleBackend for the respective experts.
  
          :param start: if True, starts server right away and returns when server is ready for requests
          :param stats_report_interval: interval between two reports of batch processing performance statistics
          :param kwargs: any other params will be forwarded to DHT upon creation
          """
      if custom_module_path is not None:
          add_custom_models_from_file(custom_module_path)
      assert expert_cls in name_to_block
  
      dht = DHT(initial_peers=initial_peers, start=True, **kwargs)
      visible_maddrs_str = [str(a) for a in dht.get_visible_maddrs()]
      logger.info(
          f"Running DHT node on {visible_maddrs_str}, initial peers = {initial_peers}"
      )
  
      assert (
          expert_pattern is None
          and num_experts is None
          and expert_uids is not None
      ) or (
          num_experts is not None and expert_uids is None
      ), "Please provide either expert_uids *or* num_experts (possibly with expert_pattern), but not both"
  
      if expert_uids is None:
          if checkpoint_dir is not None:
              assert is_directory(checkpoint_dir)
              expert_uids = [
                  child.name
                  for child in checkpoint_dir.iterdir()
                  if (child / "checkpoint_last.pt").exists()
              ]
              total_experts_in_checkpoint = len(expert_uids)
              logger.info(
                  f"Located {total_experts_in_checkpoint} checkpoints for experts {expert_uids}"
              )
  
              if total_experts_in_checkpoint > num_experts:
                  raise ValueError(
                      f"Found {total_experts_in_checkpoint} checkpoints, but num_experts is set to {num_experts}, "
                      f"which is smaller. Either increase num_experts or remove unneeded checkpoints."
                  )
          else:
              expert_uids = []
  
          uids_to_generate = num_experts - len(expert_uids)
          if uids_to_generate > 0:
              logger.info(
                  f"Generating {uids_to_generate} expert uids from pattern {expert_pattern}"
              )
              expert_uids.extend(
                  _generate_uids(uids_to_generate, expert_pattern, dht)
              )
  
      num_experts = len(expert_uids)
      num_handlers = (
          num_handlers if num_handlers is not None else num_experts * 8
      )
      device = device or (
          "cuda" if torch.cuda.is_available() else "cpu"
      )
  
      sample_input = name_to_input[expert_cls](
          DUMMY_BATCH_SIZE, hidden_dim
      )
      if isinstance(sample_input, tuple):
          args_schema = tuple(
              BatchTensorDescriptor.from_tensor(arg, compression)
              for arg in sample_input
          )
      else:
          args_schema = (
              BatchTensorDescriptor.from_tensor(
                  sample_input, compression
              ),
          )
  
      scheduler_cls = schedule_name_to_scheduler[scheduler]
      if scheduler_cls is not None:
          scheduler_cls = partial(
              scheduler_cls,
              num_warmup_steps=num_warmup_steps,
              num_training_steps=num_training_steps,
          )
  
      # initialize experts
      experts = {}
      for expert_uid in expert_uids:
          expert = name_to_block[expert_cls](hidden_dim)
          optimizer = (
              optim_cls(expert.parameters())
              if optim_cls is not None
              else None
          )
          scheduler = (
              scheduler_cls(optimizer)
              if scheduler_cls is not None
              else None
          )
          if clip_grad_norm is not None:
              optimizer = ClippingWrapper(optimizer, clip_grad_norm)
          experts[expert_uid] = ModuleBackend(
              name=expert_uid,
              module=expert,
              args_schema=args_schema,
              optimizer=optimizer,
              scheduler=scheduler,
              min_batch_size=min_batch_size,
              max_batch_size=max_batch_size,
          )
  
      if checkpoint_dir is not None:
          load_experts(experts, checkpoint_dir)
  
      return cls(
          dht,
          experts,
          num_connection_handlers=num_handlers,
          device=device,
          checkpoint_dir=checkpoint_dir,
          stats_report_interval=stats_report_interval,
          update_period=update_period,
          expiration=expiration,
          start=start,
      )
  
imports: from __future__ import annotations
  import multiprocessing
  import random
  import threading
  from contextlib import contextmanager
  from functools import partial
  from pathlib import Path
  from typing import Dict, List, Optional
  import torch
  from hivemind.dht import DHT
  from hivemind.moe.expert_uid import UID_DELIMITER
  from hivemind.moe.server.checkpoints import CheckpointSaver, is_directory, load_experts
  from hivemind.moe.server.connection_handler import ConnectionHandler
  from hivemind.moe.server.dht_handler import DHTHandlerThread, get_experts
  from hivemind.moe.server.layers import add_custom_models_from_file, name_to_block, name_to_input, schedule_name_to_scheduler
  from hivemind.moe.server.layers.optim import ClippingWrapper
  from hivemind.moe.server.module_backend import ModuleBackend
  from hivemind.moe.server.runtime import Runtime
  from hivemind.p2p import PeerInfo
  from hivemind.proto.runtime_pb2 import CompressionType
  from hivemind.utils.logging import get_logger
  from hivemind.utils.tensor_descr import DUMMY_BATCH_SIZE, BatchTensorDescriptor
label_desc: Use image node_id 2 for calling the Server obj's underlying member method code with example usage: obj.create(cls, num_experts, expert_uids, expert_pattern, expert_cls, hidden_dim, optim_cls, scheduler, num_warmup_steps, num_training_steps, clip_grad_norm, num_handlers, min_batch_size, max_batch_size, device, initial_peers, checkpoint_dir, compression, stats_report_interval, custom_module_path, update_period, expiration) and returns: cls