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 .serving import ensure_prometheus_dir

    env = {"PROMETHEUS_MULTIPROC_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 .serving import create_watcher
    from .serving import find_triton_binary

    working_dir = os.path.realpath(os.path.expanduser(working_dir))
    svc = load(bento_identifier, working_dir=working_dir)
    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),
                                *timeout_args,
                            ],
                            working_dir=working_dir,
                            numprocesses=runner.scheduled_worker_count,
                            env=env,
                        )
                    )
                    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,
                            env=env,
                        )
                    )
                    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,
            ),
        )
Experimental API for serving a BentoML runner.
src/bentoml/start.py