docstr: Runner represents a unit of computation that can be executed on a remote Python worker and scales independently
  See https://docs.bentoml.com/en/latest/concepts/runner.html for more details.
  
  Args:
      runnable_class: Runnable class that can be executed on a remote Python worker.
      runnable_init_params: Parameters to be passed to the runnable class constructor ``__init__``.
      name: Given a name for this runner. If not provided, name will be generated from the runnable class name.
            Note that all name will be converted to lowercase and validate to be a valid BentoML Tag name.
      scheduling_strategy: A strategy class that implements the scheduling logic for this runner. If not provided,
                           use the default strategy. Strategy will respect ``Runnable.SUPPORTED_RESOURCES`` as well as
                           ``Runnable.SUPPORTS_CPU_MULTI_THREADING``.
      models: An optional list composed of ``bentoml.Model`` instances.
      max_batch_size: Max batch size config for dynamic batching. If not provided, use the default value from
                      configuration.
      max_latency_ms: Max latency config for dynamic batching. If not provided, use the default value from
                      configuration.
      method_configs: A dictionary per method config for this given Runner signatures.
  
  Returns:
      :obj:`bentoml.Runner`: A Runner instance.
docstr_tok: ["Runner","represents","a","unit","of","computation","that","can","be","executed","on","a","remote","Python","worker","and","scales","independently","See","https",":","\/\/docs.bentoml.com\/en\/latest\/concepts\/runner.html","for","more","details",".","Args",":","runnable_class",":","Runnable","class","that","can","be","executed","on","a","remote","Python","worker",".","runnable_init_params",":","Parameters","to","be","passed","to","the","runnable","class","constructor","``","__init__","``",".","name",":","Given","a","name","for","this","runner",".","If","not","provided",",","name","will","be","generated","from","the","runnable","class","name",".","Note","that","all","name","will","be","converted","to","lowercase","and","validate","to","be","a","valid","BentoML","Tag","name",".","scheduling_strategy",":","A","strategy","class","that","implements","the","scheduling","logic","for","this","runner",".","If","not","provided",",","use","the","default","strategy",".","Strategy","will","respect","``","Runnable.SUPPORTED_RESOURCES","``","as","well","as","``","Runnable.SUPPORTS_CPU_MULTI_THREADING","``",".","models",":","An","optional","list","composed","of","``","bentoml.Model","``","instances",".","max_batch_size",":","Max","batch","size","config","for","dynamic","batching",".","If","not","provided",",","use","the","default","value","from","configuration",".","max_latency_ms",":","Max","latency","config","for","dynamic","batching",".","If","not","provided",",","use","the","default","value","from","configuration",".","method_configs",":","A","dictionary","per","method","config","for","this","given","Runner","signatures",".","Returns",":",":","obj",":","`","bentoml.Runner","`",":","A","Runner","instance","."]
code: def __init__(
      self,
      runnable_class: type[Runnable],
      *,
      runnable_init_params: dict[str, t.Any] | None = None,
      name: str | None = None,
      scheduling_strategy: type[Strategy] = DefaultStrategy,
      models: list[Model] | None = None,
      max_batch_size: int | None = None,
      max_latency_ms: int | None = None,
      method_configs: dict[str, dict[str, int]] | None = None,
      embedded: bool = False,
  ) -> None:
      """
  
      Runner represents a unit of computation that can be executed on a remote Python worker and scales independently
      See https://docs.bentoml.com/en/latest/concepts/runner.html for more details.
  
      Args:
          runnable_class: Runnable class that can be executed on a remote Python worker.
          runnable_init_params: Parameters to be passed to the runnable class constructor ``__init__``.
          name: Given a name for this runner. If not provided, name will be generated from the runnable class name.
                Note that all name will be converted to lowercase and validate to be a valid BentoML Tag name.
          scheduling_strategy: A strategy class that implements the scheduling logic for this runner. If not provided,
                               use the default strategy. Strategy will respect ``Runnable.SUPPORTED_RESOURCES`` as well as
                               ``Runnable.SUPPORTS_CPU_MULTI_THREADING``.
          models: An optional list composed of ``bentoml.Model`` instances.
          max_batch_size: Max batch size config for dynamic batching. If not provided, use the default value from
                          configuration.
          max_latency_ms: Max latency config for dynamic batching. If not provided, use the default value from
                          configuration.
          method_configs: A dictionary per method config for this given Runner signatures.
  
      Returns:
          :obj:`bentoml.Runner`: A Runner instance.
      """
  
      if name is None:
          name = runnable_class.__name__.lower()
          logger.warning(
              "Using lowercased runnable class name '%s' for runner.",
              name,
          )
  
      runners_config = BentoMLContainer.config.runners.get()
      # If given runner is configured, then use it. Otherwise use the default configuration.
      if name in runners_config:
          config = runners_config[name]
      else:
          config = runners_config
  
      if models is None:
          models = []
  
      runner_method_map: dict[
          str, RunnerMethod[t.Any, t.Any, t.Any]
      ] = {}
      method_configs = {} if method_configs is None else method_configs
  
      if runnable_class.bentoml_runnable_methods__ is None:
          raise ValueError(
              f"Runnable class '{runnable_class.__name__}' has no methods!"
          )
  
      for (
          method_name,
          method,
      ) in runnable_class.bentoml_runnable_methods__.items():
          if not config["batching"]["enabled"]:
              method.config.batchable = False
  
          method_max_batch_size = None
          method_max_latency_ms = None
          if method_name in method_configs:
              method_max_batch_size = method_configs[method_name].get(
                  "max_batch_size"
              )
              method_max_latency_ms = method_configs[method_name].get(
                  "max_latency_ms"
              )
  
          runner_method_map[method_name] = RunnerMethod(
              runner=self,
              name=method_name,
              config=method.config,
              max_batch_size=first_not_none(
                  method_max_batch_size,
                  max_batch_size,
                  default=config["batching"]["max_batch_size"],
              ),
              max_latency_ms=first_not_none(
                  method_max_latency_ms,
                  max_latency_ms,
                  default=config["batching"]["max_latency_ms"],
              ),
          )
  
      self.__attrs_init__(
          name=name,
          models=models,
          runnable_class=runnable_class,
          runnable_init_params=runnable_init_params,
          resource_config=config["resources"],
          workers_per_resource=config.get("workers_per_resource", 1),
          runner_methods=list(runner_method_map.values()),
          scheduling_strategy=scheduling_strategy,
          embedded=embedded,
      )
  
      # Choose the default method:
      #  1. if there's only one method, it will be set as default
      #  2. if there's a method named "__call__", it will be set as default
      #  3. otherwise, there's no default method
      if len(runner_method_map) == 1:
          default_method = next(iter(runner_method_map.values()))
          logger.debug(
              "Default runner method set to '%s', it can be accessed both via 'runner.run' and 'runner.%s.async_run'.",
              default_method.name,
              default_method.name,
          )
      elif "__call__" in runner_method_map:
          default_method = runner_method_map["__call__"]
          logger.debug(
              "Default runner method set to '__call__', it can be accessed via 'runner.run' or 'runner.async_run'."
          )
      else:
          default_method = None
          logger.debug(
              "No default method found for Runner '%s', all method access needs to be in the form of 'runner.{method}.run'.",
              name,
          )
  
      # set default run method entrypoint
      if default_method is not None:
          object.__setattr__(self, "run", default_method.run)
          object.__setattr__(
              self, "async_run", default_method.async_run
          )
  
      # set all run method entrypoint
      for runner_method in self.runner_methods:
          object.__setattr__(self, runner_method.name, runner_method)
  
imports: from __future__ import annotations
  import logging
  import typing
  from abc import ABC
  from abc import abstractmethod
  import attr
  from simple_di import Provide
  from simple_di import inject
  from ...exceptions import StateException
  from ..configuration.containers import BentoMLContainer
  from ..models.model import Model
  from ..tag import validate_tag_str
  from ..utils import first_not_none
  from .runnable import Runnable
  from .runner_handle import DummyRunnerHandle
  from .runner_handle import RunnerHandle
  from .strategy import DefaultStrategy
  from .strategy import Strategy
label_desc: Use image node_id 2 to create a new Runner object from inherited base classes: AbstractRunner with example: obj = Runner(runnable_class)