code: def start_runner_server(
      bento_identifier: str,
      working_dir: str,
      runner_name: str,
      port: int | None = None,
      host: str | None = None,
      timeout: int | None = None,
      backlog: int = Provide[
          BentoMLContainer.api_server_config.backlog
      ],
  ) -> None:
      """
      Experimental API for serving a BentoML runner.
      """
      from .serve import ensure_prometheus_dir
  
      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 .serve import create_watcher
      from .serve import find_triton_binary
  
      working_dir = os.path.realpath(os.path.expanduser(working_dir))
      svc = load(
          bento_identifier,
          working_dir=working_dir,
          standalone_load=True,
      )
      timeout_args = ["--timeout", str(timeout)] if timeout else []
  
      from circus.sockets import CircusSocket  # type: ignore
      from circus.watcher import Watcher  # type: ignore
  
      watchers: list[Watcher] = []
      circus_socket_map: dict[str, CircusSocket] = {}
  
      # NOTE: We need to find and set model-repository args
      # to all TritonRunner instances (required from tritonserver if spawning multiple instances.)
  
      with contextlib.ExitStack() as port_stack:
          for runner in svc.runners:
              if runner.name == runner_name:
                  if isinstance(runner, Runner):
                      if port is None:
                          port = port_stack.enter_context(
                              reserve_free_port()
                          )
                      if host is None:
                          host = "127.0.0.1"
                      circus_socket_map[runner.name] = CircusSocket(
                          name=runner.name,
                          host=host,
                          port=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,
                                  *timeout_args,
                              ],
                              working_dir=working_dir,
                              numprocesses=runner.scheduled_worker_count,
                          )
                      )
                      break
                  else:
                      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,
                          )
                      )
                      break
          else:
              raise ValueError(
                  f"Runner {runner_name} not found in the service: `{bento_identifier}`, "
                  f"available runners: {[r.name for r in svc.runners]}"
              )
  
      arbiter = create_standalone_arbiter(
          watchers=watchers,
          sockets=list(circus_socket_map.values()),
      )
      with track_serve(svc, production=True, component=RUNNER):
          arbiter.start(
              cb=lambda _: logger.info(  # type: ignore
                  'Starting RunnerServer from "%s" running on http://%s:%s (Press CTRL+C to quit)',
                  bento_identifier,
                  host,
                  port,
              ),
          )
  
file_path: BentoML/grpc-client/bentoml/start.py
imports: from __future__ import annotations
  import contextlib
  import json
  import logging
  import os
  from simple_di import Provide
  from simple_di import inject
  from ._internal.configuration.containers import BentoMLContainer
  from ._internal.runner.runner import Runner
  from .grpc.utils import LATEST_PROTOCOL_VERSION
label_desc: Use image node_id 1 for calling a global function with example usage: start_runner_server(bento_identifier, working_dir, runner_name, port, host, timeout, backlog) without return types