code: def serve_grpc_production(
      bento_identifier: str,
      working_dir: str,
      port: int = Provide[BentoMLContainer.grpc.port],
      host: str = Provide[BentoMLContainer.grpc.host],
      bentoml_home: str = Provide[BentoMLContainer.bentoml_home],
      backlog: int = Provide[
          BentoMLContainer.api_server_config.backlog
      ],
      api_workers: int = Provide[BentoMLContainer.api_server_workers],
      ssl_certfile: str | None = Provide[BentoMLContainer.ssl.certfile],
      ssl_keyfile: str | None = Provide[BentoMLContainer.ssl.keyfile],
      ssl_ca_certs: str | None = Provide[BentoMLContainer.ssl.ca_certs],
      max_concurrent_streams: int
      | None = Provide[BentoMLContainer.grpc.max_concurrent_streams],
      channelz: bool = Provide[BentoMLContainer.grpc.channelz.enabled],
      reflection: bool = Provide[
          BentoMLContainer.grpc.reflection.enabled
      ],
      protocol_version: str = LATEST_PROTOCOL_VERSION,
      reload: bool = False,
      development_mode: bool = False,
  ) -> None:
      prometheus_dir = ensure_prometheus_dir()
  
      from . import load
      from ._internal.utils import reserve_free_port
      from ._internal.utils.analytics import track_serve
      from ._internal.utils.circus import create_standalone_arbiter
      from ._internal.utils.uri import path_to_uri
  
      working_dir = os.path.realpath(os.path.expanduser(working_dir))
      svc = load(bento_identifier, working_dir=working_dir)
  
      from circus.sockets import CircusSocket  # type: ignore
  
      watchers: list[Watcher] = []
      circus_socket_map: dict[str, CircusSocket] = {}
      runner_bind_map: dict[str, str] = {}
      uds_path = None
  
      # Check whether users are running --grpc on windows
      # also raising warning if users running on MacOS or FreeBSD
      if psutil.WINDOWS and (not development_mode):
          raise BentoMLException(
              "'grpc' is not supported on Windows without '--development'. The reason being SO_REUSEPORT socket option is only available on UNIX system, and gRPC implementation depends on this behaviour."
          )
      if psutil.MACOS or psutil.FREEBSD:
          logger.warning(
              "Due to gRPC implementation on exposing SO_REUSEPORT, BentoML production server's behaviour on %s is not correct. We recommend to containerize BentoServer as a Linux container instead. For testing locally, use `bentoml serve --development`",
              "MacOS" if psutil.MACOS else "FreeBSD",
          )
  
      # NOTE: We need to find and set model-repository args
      # to all TritonRunner instances (required from tritonserver if spawning multiple instances.)
  
      if psutil.POSIX and not IS_WSL:
          # use AF_UNIX sockets for Circus
          uds_path = tempfile.mkdtemp()
          for runner in svc.runners:
              if isinstance(runner, Runner):
                  if runner.embedded or development_mode:
                      continue
  
                  sockets_path = os.path.join(
                      uds_path, f"{id(runner)}.sock"
                  )
                  assert len(sockets_path) < MAX_AF_UNIX_PATH_LENGTH
  
                  runner_bind_map[runner.name] = path_to_uri(
                      sockets_path
                  )
                  circus_socket_map[runner.name] = CircusSocket(
                      name=runner.name,
                      path=sockets_path,
                      backlog=backlog,
                  )
  
                  watchers.append(
                      create_watcher(
                          name=f"runner_{runner.name}",
                          args=[
                              "-m",
                              SCRIPT_RUNNER,
                              bento_identifier,
                              "--runner-name",
                              runner.name,
                              "--fd",
                              f"$(circus.sockets.{runner.name})",
                              "--working-dir",
                              working_dir,
                              "--worker-id",
                              "$(CIRCUS.WID)",
                              "--worker-env-map",
                              json.dumps(
                                  runner.scheduled_worker_env_map
                              ),
                          ],
                          working_dir=working_dir,
                          numprocesses=runner.scheduled_worker_count,
                      )
                  )
              else:
                  # Make sure that the tritonserver uses the correct protocol
                  runner_bind_map[runner.name] = runner.protocol_address
                  cli_args = runner.cli_args + [
                      f"--http-port={runner.protocol_address.split(':')[-1]}"
                      if runner.tritonserver_type == "http"
                      else f"--grpc-port={runner.protocol_address.split(':')[-1]}"
                  ]
                  watchers.append(
                      create_watcher(
                          name=f"tritonserver_{runner.name}",
                          cmd=find_triton_binary(),
                          args=cli_args,
                          use_sockets=False,
                          working_dir=working_dir,
                          numprocesses=1,
                      )
                  )
  
      elif psutil.WINDOWS or IS_WSL:
          # Windows doesn't (fully) support AF_UNIX sockets
          with contextlib.ExitStack() as port_stack:
              for runner in svc.runners:
                  if isinstance(runner, Runner):
                      if runner.embedded or development_mode:
                          continue
  
                      runner_port = port_stack.enter_context(
                          reserve_free_port()
                      )
                      runner_host = "127.0.0.1"
  
                      runner_bind_map[
                          runner.name
                      ] = f"tcp://{runner_host}:{runner_port}"
                      circus_socket_map[runner.name] = CircusSocket(
                          name=runner.name,
                          host=runner_host,
                          port=runner_port,
                          backlog=backlog,
                      )
  
                      watchers.append(
                          create_watcher(
                              name=f"runner_{runner.name}",
                              args=[
                                  "-m",
                                  SCRIPT_RUNNER,
                                  bento_identifier,
                                  "--runner-name",
                                  runner.name,
                                  "--fd",
                                  f"$(circus.sockets.{runner.name})",
                                  "--working-dir",
                                  working_dir,
                                  "--no-access-log",
                                  "--worker-id",
                                  "$(circus.wid)",
                                  "--worker-env-map",
                                  json.dumps(
                                      runner.scheduled_worker_env_map
                                  ),
                                  "--prometheus-dir",
                                  prometheus_dir,
                              ],
                              working_dir=working_dir,
                              numprocesses=runner.scheduled_worker_count,
                          )
                      )
                  else:
                      # Make sure that the tritonserver uses the correct protocol
                      runner_bind_map[
                          runner.name
                      ] = runner.protocol_address
                      cli_args = runner.cli_args + [
                          f"--http-port={runner.protocol_address.split(':')[-1]}"
                          if runner.tritonserver_type == "http"
                          else f"--grpc-port={runner.protocol_address.split(':')[-1]}"
                      ]
                      watchers.append(
                          create_watcher(
                              name=f"tritonserver_{runner.name}",
                              cmd=find_triton_binary(),
                              args=cli_args,
                              use_sockets=False,
                              working_dir=working_dir,
                              numprocesses=1,
                          )
                      )
              # reserve one more to avoid conflicts
              port_stack.enter_context(reserve_free_port())
      else:
          raise NotImplementedError(
              "Unsupported platform: {}".format(sys.platform)
          )
  
      logger.debug("Runner map: %s", runner_bind_map)
  
      ssl_args = construct_ssl_args(
          ssl_certfile=ssl_certfile,
          ssl_keyfile=ssl_keyfile,
          ssl_ca_certs=ssl_ca_certs,
      )
      scheme = "https" if BentoMLContainer.ssl.enabled.get() else "http"
  
      close_child_stdin: bool = False if development_mode else True
  
      loop = asyncio.get_event_loop()
      loop.run_until_complete(on_service_deployment(svc))
  
      with contextlib.ExitStack() as port_stack:
          api_port = port_stack.enter_context(
              reserve_free_port(
                  host, port=port, enable_so_reuseport=True
              )
          )
          api_server_args = [
              "-m",
              SCRIPT_GRPC_API_SERVER,
              bento_identifier,
              "--host",
              host,
              "--port",
              str(api_port),
              "--runner-map",
              json.dumps(runner_bind_map),
              "--working-dir",
              working_dir,
              "--prometheus-dir",
              prometheus_dir,
              "--worker-id",
              "$(CIRCUS.WID)",
              *ssl_args,
              "--protocol-version",
              protocol_version,
          ]
  
          if reflection:
              api_server_args.append("--enable-reflection")
          if channelz:
              api_server_args.append("--enable-channelz")
          if max_concurrent_streams:
              api_server_args.extend(
                  [
                      "--max-concurrent-streams",
                      str(max_concurrent_streams),
                  ]
              )
  
          if development_mode:
              api_server_args.append("--development-mode")
  
          watchers.append(
              create_watcher(
                  name="grpc_api_server",
                  args=api_server_args,
                  use_sockets=False,
                  working_dir=working_dir,
                  numprocesses=api_workers,
                  close_child_stdin=close_child_stdin,
              )
          )
  
      if BentoMLContainer.api_server_config.metrics.enabled.get():
          metrics_host = BentoMLContainer.grpc.metrics.host.get()
          metrics_port = BentoMLContainer.grpc.metrics.port.get()
  
          circus_socket_map[PROMETHEUS_SERVER_NAME] = CircusSocket(
              name=PROMETHEUS_SERVER_NAME,
              host=metrics_host,
              port=metrics_port,
              backlog=backlog,
          )
  
          watchers.append(
              create_watcher(
                  name="prom_server",
                  args=[
                      "-m",
                      SCRIPT_GRPC_PROMETHEUS_SERVER,
                      "--fd",
                      f"$(circus.sockets.{PROMETHEUS_SERVER_NAME})",
                      "--prometheus-dir",
                      prometheus_dir,
                      "--backlog",
                      f"{backlog}",
                  ],
                  working_dir=working_dir,
                  numprocesses=1,
                  singleton=True,
                  close_child_stdin=close_child_stdin,
              )
          )
  
          log_metrics_host = (
              "127.0.0.1" if metrics_host == "0.0.0.0" else metrics_host
          )
  
          logger.info(
              PROMETHEUS_MESSAGE,
              "gRPC",
              bento_identifier,
              f"http://{log_metrics_host}:{metrics_port}",
          )
  
      arbiter_kwargs: dict[str, t.Any] = {
          "watchers": watchers,
          "sockets": list(circus_socket_map.values()),
      }
  
      plugins = []
  
      if reload:
          reload_plugin = make_reload_plugin(working_dir, bentoml_home)
          plugins.append(reload_plugin)
  
      arbiter_kwargs["plugins"] = plugins
  
      if development_mode:
          arbiter_kwargs["debug"] = (
              True if sys.platform != "win32" else False
          )
          arbiter_kwargs["loggerconfig"] = SERVER_LOGGING_CONFIG
          arbiter_kwargs["loglevel"] = "WARNING"
  
      arbiter = create_standalone_arbiter(**arbiter_kwargs)
  
      production: bool = False if development_mode else True
      with track_serve(svc, production=production, serve_kind="grpc"):
          try:
              arbiter.start(
                  cb=lambda _: logger.info(  # type: ignore
                      'Starting production %s BentoServer from "%s" listening on %s://%s:%d (Press CTRL+C to quit)',
                      "gRPC",
                      bento_identifier,
                      scheme,
                      host,
                      port,
                  ),
              )
          finally:
              if uds_path is not None:
                  shutil.rmtree(uds_path)
  
imports: from __future__ import annotations
  import asyncio
  import contextlib
  import json
  import logging
  import os
  import platform
  import shlex
  import shutil
  import sys
  import tempfile
  import typing
  from functools import partial
  from pathlib import Path
  import psutil
  from simple_di import Provide
  from simple_di import inject
  from bentoml._internal.log import SERVER_LOGGING_CONFIG
  from ._internal.configuration.containers import BentoMLContainer
  from ._internal.runner.runner import Runner
  from ._internal.utils import is_async_callable
  from .exceptions import BentoMLConfigException
  from .exceptions import BentoMLException
  from .grpc.utils import LATEST_PROTOCOL_VERSION
label_desc: Use image node_id 9 for calling a global function with example usage: serve_grpc_production(bento_identifier, working_dir, port, host, bentoml_home, backlog, api_workers, ssl_certfile, ssl_keyfile, ssl_ca_certs, max_concurrent_streams, channelz, reflection, protocol_version, reload, development_mode) without return types