class_docstr: A decorator that:
      * wrap batch function
      * implement CORK algorithm to cork & release calling of wrapped function
  The wrapped function should be an async function.
class_docstr_tok: ["A","decorator","that",":","*","wrap","batch","function","*","implement","CORK","algorithm","to","cork","&","release","calling","of","wrapped","function","The","wrapped","function","should","be","an","async","function","."]
docstr: params:
      * max_latency_in_ms: max_latency_in_ms for inbound tasks in milliseconds
      * max_batch_size: max batch size of inbound tasks
      * shared_sema: semaphore to limit concurrent outbound tasks
      * fallback: callable to return fallback result
  raises:
      * all possible exceptions the decorated function has
docstr_tok: ["params",":","*","max_latency_in_ms",":","max_latency_in_ms","for","inbound","tasks","in","milliseconds","*","max_batch_size",":","max","batch","size","of","inbound","tasks","*","shared_sema",":","semaphore","to","limit","concurrent","outbound","tasks","*","fallback",":","callable","to","return","fallback","result","raises",":","*","all","possible","exceptions","the","decorated","function","has"]
code: def __init__(
      self,
      max_latency_in_ms: int,
      max_batch_size: int,
      shared_sema: t.Optional[NonBlockSema] = None,
      fallback: t.Callable[[], t.Any] | type[t.Any] | None = None,
  ):
      """
      params:
          * max_latency_in_ms: max_latency_in_ms for inbound tasks in milliseconds
          * max_batch_size: max batch size of inbound tasks
          * shared_sema: semaphore to limit concurrent outbound tasks
          * fallback: callable to return fallback result
      raises:
          * all possible exceptions the decorated function has
      """
      self.max_latency = max_latency_in_ms / 1000.0
      self.fallback = fallback
      self.optimizer = Optimizer(self.max_latency)
      self.max_batch_size = int(max_batch_size)
      self.tick_interval = 0.001
  
      self._controller = None
      self._queue: collections.deque[
          Job
      ] = collections.deque()  # TODO(bojiang): maxlen
      self._sema = shared_sema if shared_sema else NonBlockSema(1)
  
imports: from __future__ import annotations
  import asyncio
  import collections
  import functools
  import logging
  import time
  import traceback
  import typing
  from functools import cached_property
  import attr
  import numpy
  from ..utils.alg import TokenBucket
label_desc: Use image node_id 1 to create a new CorkDispatcher object  with example: obj = CorkDispatcher(max_latency_in_ms, max_batch_size, shared_sema, fallback)