class_docstr: A group of processes that processes incoming requests for multiple module backends on a shared device.
  Runtime is usually created and managed by Server, humans need not apply.
  
  For debugging, you can start runtime manually with .start() or .run()
  
  >>> module_backends = {'expert_name': ModuleBackend(**kwargs)}
  >>> runtime = Runtime(module_backends)
  >>> runtime.start()  # start runtime in background thread. To start in current thread, use runtime.run()
  >>> runtime.ready.wait()  # await for runtime to load all experts on device and create request pools
  >>> future = runtime.module_backends['expert_name'].forward_pool.submit_task(*module_inputs)
  >>> print("Returned:", future.result())
  >>> runtime.shutdown()
  
  :param module_backends: a dict [expert uid -> ModuleBackend]
  :param prefetch_batches: form up to this many batches in advance
  :param sender_threads: dispatches outputs from finished batches using this many asynchronous threads
  :param device: if specified, moves all experts and data to this device via .to(device=device).
    If you want to manually specify devices for each expert (in their forward pass), leave device=None (default)
  
  :param stats_report_interval: interval to collect and log statistics about runtime performance
class_docstr_tok: ["A","group","of","processes","that","processes","incoming","requests","for","multiple","module","backends","on","a","shared","device",".","Runtime","is","usually","created","and","managed","by","Server",",","humans","need","not","apply",".","For","debugging",",","you","can","start","runtime","manually","with",".start","(",")","or",".run","(",")",">",">",">","module_backends","=","{","'expert_name","'",":","ModuleBackend","(","*","*","kwargs",")","}",">",">",">","runtime","=","Runtime","(","module_backends",")",">",">",">","runtime.start","(",")","#","start","runtime","in","background","thread",".","To","start","in","current","thread",",","use","runtime.run","(",")",">",">",">","runtime.ready.wait","(",")","#","await","for","runtime","to","load","all","experts","on","device","and","create","request","pools",">",">",">","future","=","runtime.module_backends","[","'expert_name","'","]",".forward_pool.submit_task","(","*","module_inputs",")",">",">",">","print","(","``","Returned",":","''",",","future.result","(",")",")",">",">",">","runtime.shutdown","(",")",":","param","module_backends",":","a","dict","[","expert","uid","-",">","ModuleBackend","]",":","param","prefetch_batches",":","form","up","to","this","many","batches","in","advance",":","param","sender_threads",":","dispatches","outputs","from","finished","batches","using","this","many","asynchronous","threads",":","param","device",":","if","specified",",","moves","all","experts","and","data","to","this","device","via",".to","(","device=device",")",".","If","you","want","to","manually","specify","devices","for","each","expert","(","in","their","forward","pass",")",",","leave","device=None","(","default",")",":","param","stats_report_interval",":","interval","to","collect","and","log","statistics","about","runtime","performance"]
code: def run(self):
      for pool in self.pools:
          if not pool.is_alive():
              pool.start()
      if self.device is not None:
          for backend in self.module_backends.values():
              backend.module.to(self.device)
  
      with mp.pool.ThreadPool(
          self.sender_threads
      ) as output_sender_pool:
          try:
              self.ready.set()
              if self.stats_report_interval is not None:
                  self.stats_reporter.start()
              logger.info("Started")
  
              batch_iterator = self.iterate_minibatches_from_pools()
              if self.prefetch_batches > 0:
                  batch_iterator = BackgroundGenerator(
                      batch_iterator, self.prefetch_batches
                  )
  
              for pool, batch_index, batch in batch_iterator:
                  logger.debug(
                      f"Processing batch {batch_index} from pool {pool.name}"
                  )
                  start = perf_counter()
                  try:
                      outputs, batch_size = self.process_batch(
                          pool, batch_index, *batch
                      )
                      batch_processing_time = perf_counter() - start
                      logger.debug(
                          f"Pool {pool.name}: batch {batch_index} processed, size {batch_size}"
                      )
                      if self.stats_report_interval is not None:
                          self.stats_reporter.report_stats(
                              pool.name,
                              batch_size,
                              batch_processing_time,
                          )
  
                      output_sender_pool.apply_async(
                          pool.send_outputs_from_runtime,
                          args=[batch_index, outputs],
                      )
                  except KeyboardInterrupt:
                      raise
                  except BaseException as exception:
                      logger.exception(
                          f"Caught {exception}, attempting to recover"
                      )
                      output_sender_pool.apply_async(
                          pool.send_exception_from_runtime,
                          args=[batch_index, exception],
                      )
  
          finally:
              if not self.shutdown_trigger.is_set():
                  self.shutdown()
  
imports: import multiprocessing
  import multiprocessing.pool
  import threading
  from collections import defaultdict
  from itertools import chain
  from queue import SimpleQueue
  from selectors import EVENT_READ, DefaultSelector
  from statistics import mean
  from time import perf_counter
  from typing import Any, Dict, NamedTuple, Optional, Tuple
  import torch
  from prefetch_generator import BackgroundGenerator
  from hivemind.moe.server.module_backend import ModuleBackend
  from hivemind.moe.server.task_pool import TaskPoolBase
  from hivemind.utils import get_logger
label_desc: Use image node_id 2 for calling the Runtime obj's underlying member method code with example usage: obj.run() without return types