class_docstr: An async implementation of a gRPC server.
code: def run(self) -> None:
      try:
          self.loop.run_until_complete(self.serve())
      finally:
          try:
              self.loop.call_soon_threadsafe(
                  lambda: asyncio.ensure_future(self.shutdown())
              )
          except Exception as e:  # pylint: disable=broad-except
              raise RuntimeError(
                  f"Server failed unexpectedly: {e}"
              ) from None
  
imports: from __future__ import annotations
  import asyncio
  import inspect
  import logging
  import os
  import sys
  import typing
  from concurrent.futures import ThreadPoolExecutor
  from functools import cached_property
  from functools import partial
  from typing import TYPE_CHECKING
  from simple_di import Provide
  from simple_di import inject
  from ...grpc.utils import LATEST_PROTOCOL_VERSION
  from ...grpc.utils import import_generated_stubs
  from ...grpc.utils import import_grpc
  from ...grpc.utils import load_from_file
  from ..configuration.containers import BentoMLContainer
  from ..context import ServiceContext
  from ..utils import LazyLoader
label_desc: Use image node_id 6 for calling the Server obj's underlying member method code with example usage: obj.run() without return types