Repository: jina-ai/serve
Pull request: feat: dynamic batching
allow dynamic batching for executors. Feature described in detail [here](https://github.com/jina-ai/jina/issues/5033).

## Dynamic Batching Constraints
- len(input) == len(output)
- merge must reduce docs (no docs_matrix)
- no dictionary return allowed

## To Do

- [x] create the decorator
- [x] make overridable in YAML
- [x] implement queue and batching in the worker
- [x] implement unit tests for batch queue
- [x] test that failures are propagated
- [x] make sure default endpoint is covered
- [x] allow endpoint specific configs
- [x] add support for parameters
- [x] ~~make max_batch_size work~~ removed afterwards
- [x] test endpoint specific configs
- [x] remove `max_batch_size`; this could be a useful addition for future iterations OR do it properly, splitting up individual requests that are longer than the max
- [x] verify user experience with `Executor.serve()`
- [x] handle sigterms
- [x] add more tests: 
* no dynamic_batching decorator, only specified in the yaml
* dynamic_batching decorator + yaml
* serve.reload + dynamic batching
- [x] what happens when request is received and the task is being flushed ? fixed in beec7cc08e18c3cd8bbe68bfa485d691e8abf27b
- [x] documentation


Do we still need this?
- implement waiting logic in the gateway
- fix tracing in executor call
  - It seems like the tracing works just fine even when we dont pass the context into the executor call.
  - Just the nesting levels are different.
  - Do we need to link spans here? @girishc13 
- make sure teardown flushes
  - this is not helpful because the grpc requests are all cancelled before teardown is called
  - this is true even without dynamic batching -- any executors executing halfway have their operations cancelled

**Example Usage**
As decorator:
```python
class MyExecutor(Executor):
    @requests(on=['/foo'])
    @dynamic_batching(preferred_batch_size=4, timeout=1000)
    def foo_fun(self, docs, **kwargs):
        for doc in docs:
            doc.text += FOO_SUCCESS_MSG
```
You can override this when adding to the flow:
```python
f = Flow(port=12345).add(
        uses=MyExecutor,
        name='exec0',
        uses_dynamic_batching={'/foo': {'preferred_batch_size': 2, 'timeout': 4000}},
    )
```
Or in the YAML:
```yaml
! MyExecutor
py_modules:
  - my_executor.py
dynamic_batching:
  /foo:
    preferred_batch_size: 2
    timeout: 4000
  /bar:
    preferred_batch_size: 4
    timeout: 2000
```


Problem statement:
Allow dynamic batching in Executor
**Describe the feature**
We have seen that the simple `Request-Response` model with a static `request_size` is limiting in some cases. For instance some `Encoders` would rather accumulate `requests` until a big `enough` batch is accumulated and then all encoded at once. This could be used to improve `throughput` at a potential cost of `latency`.


How can I imagine the Feature:

In `executor.py`:

```python
from jina import Executor, requests, dynamic_batch

class DynamicBatchExecutor(Executor):

    @requests
    @dynamic_batch(batch_size=...) # potentially other arguments
     def foo(self, docs, **kwargs): 
           # decorator makes sure X docs arrive here (X being batch size)
           pass
```

Should the `Executor` do it or allowed to be passed dynamically in the `Runtime`? To be decided as it is more of a UI issue

This would affect the `Req-Response` model in the gateway. How I would solve it without affecting the `Request-Response` model from the client?

- [ ] In our proto, have some field in the `Response` stating that Response is yet to be served.
- [ ] The gateway keeps a map with the `request_id` and the `doc_ids` present there. 
- [ ] The gateway waits for the next Response that is served from that node, and then fires the continuation of all the waiting `Requests` with the appropiate `Documents` attached.

This way, we achieve.

- [ ] Dynamic batching
- [ ] The rest of the Flow and Executors do not realize about this. 
- [ ] The interface with the Client is kept. Single `request-response` model is kept from user perspective

Small hidden implementation details:
- How to deal with the last hanging docs.
- Imagine I have dynamic_batch_size of 50 and I just send 102 Documents. How to handle the 2 remaining Docs? I just have a timeout or something?

Hints from issue discussion:
@JoanFM @JohannesMessner hey!
I am very interested in this feature. Can you tell, what is the progress here?

Also, can you add `timeout` parameter for `@dynamic_batch` which tells Executer to process all pending requests if some of them is older than `timeout` and their amount lower than `batch_size`?
I am currently finishing off some other work and will start actively working on this feature this week.
And yes, we will most likely have a mechanism that allows non-complete batches to be processed, a `timeout` seems reasonable.

@mister-fahrenheit do you have any other requirements or wishes for this feature? Anything about your use case that you would want to share? We are always happy to get inspiration from our users for features like this!
@JohannesMessner glad to hear it!

I am building an app where the users send many requests, but only with a single document. And Executor works really faster if we pass `n` documents in batch instead of one by one, even with some latency. So, this feature should help us a lot. 
Ok perfect, that is exactly what we have in mind and should be nicely covered by this feature. I'll keep you posted on the progress!

**Update:** Implementation started [here](https://github.com/jina-ai/jina/pull/5340).

Commit message: test: create folder for batching tests

Signed-off-by: Johannes Messner <messnerjo@gmail.com>

Patch to review:
diff --git a/docs/fundamentals/flow/executor-args.md b/docs/fundamentals/flow/executor-args.md
index a2ec046e01879..0b11e37ed48f8 100644
--- a/docs/fundamentals/flow/executor-args.md
+++ b/docs/fundamentals/flow/executor-args.md
@@ -14,6 +14,7 @@
 | `uses_with` | Dictionary of keyword arguments that will override the `with` configuration in `uses` | `object` | `None` |
 | `uses_metas` | Dictionary of keyword arguments that will override the `metas` configuration in `uses` | `object` | `None` |
 | `uses_requests` | Dictionary of keyword arguments that will override the `requests` configuration in `uses` | `object` | `None` |
+| `uses_dynamic_batching` | Dictionary of keyword arguments that will override the `dynamic_batching` configuration in `uses` | `object` | `None` |
 | `py_modules` | The customized python modules need to be imported before loading the executor<br><br>Note that the recommended way is to only import a single module - a simple python file, if your<br>executor can be defined in a single file, or an ``__init__.py`` file if you have multiple files,<br>which should be structured as a python package. For more details, please see the<br>`Executor cookbook <https://docs.jina.ai/fundamentals/executor/executor-files/>`__ | `array` | `None` |
 | `output_array_type` | The type of array `tensor` and `embedding` will be serialized to.<br><br>Supports the same types as `docarray.to_protobuf(.., ndarray_type=...)`, which can be found <br>`here <https://docarray.jina.ai/fundamentals/document/serialization/#from-to-protobuf>`.<br>Defaults to retaining whatever type is returned by the Executor. | `string` | `None` |
 | `exit_on_exceptions` | List of exceptions that will cause the Executor to shut down. | `array` | `[]` |
diff --git a/jina/__init__.py b/jina/__init__.py
index ef8f12482bc09..a52999f4ac8a5 100644
--- a/jina/__init__.py
+++ b/jina/__init__.py
@@ -211,7 +211,7 @@ def _set_nofile(nofile_atleast=4096):
 
 # Executor
 from jina.serve.executors import BaseExecutor as Executor
-from jina.serve.executors.decorators import monitor, requests
+from jina.serve.executors.decorators import dynamic_batching, monitor, requests
 
 # Custom Gateway
 from jina.serve.gateway import BaseGateway as Gateway
diff --git a/jina/jaml/__init__.py b/jina/jaml/__init__.py
index 58a08bbed6330..a87f79cf9773b 100644
--- a/jina/jaml/__init__.py
+++ b/jina/jaml/__init__.py
@@ -615,6 +615,7 @@ def load_config(
         extra_search_paths: Optional[List[str]] = None,
         py_modules: Optional[str] = None,
         runtime_args: Optional[Dict[str, Any]] = None,
+        uses_dynamic_batching: Optional[Dict] = None,
         **kwargs,
     ) -> 'JAMLCompatible':
         """A high-level interface for loading configuration with features
@@ -667,7 +668,7 @@ def load_config(
         :param extra_search_paths: extra paths used when looking for executor yaml files
         :param py_modules: Optional py_module from which the object need to be loaded
         :param runtime_args: Optional dictionary of parameters runtime_args to be directly passed without being parsed into a yaml config
-        :param : runtime_args that need to be passed to the yaml
+        :param uses_dynamic_batching: dictionary of parameters to overwrite from the default config's dynamic_batching field
 
         :param kwargs: kwargs for parse_config_source
         :return: :class:`JAMLCompatible` object
@@ -713,9 +714,12 @@ def _delitem(
                     _delitem(no_tag_yml, key='uses_metas')
                 if uses_requests is not None:
                     _delitem(no_tag_yml, key='uses_requests')
+                if uses_dynamic_batching is not None:
+                    _delitem(no_tag_yml, key='uses_dynamic_batching')
                 cls._override_yml_params(no_tag_yml, 'with', uses_with)
                 cls._override_yml_params(no_tag_yml, 'metas', uses_metas)
                 cls._override_yml_params(no_tag_yml, 'requests', uses_requests)
+                cls._override_yml_params(no_tag_yml, 'dynamic_batching', uses_dynamic_batching)
 
             else:
                 raise BadConfigSource(
diff --git a/jina/jaml/parsers/executor/legacy.py b/jina/jaml/parsers/executor/legacy.py
index efcb0527d55b7..4d61fe9c3abbf 100644
--- a/jina/jaml/parsers/executor/legacy.py
+++ b/jina/jaml/parsers/executor/legacy.py
@@ -39,6 +39,7 @@ def parse(
                 **data.get('with', {}),
                 metas=data.get('metas', {}),
                 requests=data.get('requests', {}),
+                dynamic_batching=data.get('dynamic_batching', {}),
                 runtime_args=runtime_args,
             )
         else:
@@ -46,6 +47,7 @@ def parse(
                 **data.get('with', {}),
                 metas=data.get('metas', {}),
                 requests=data.get('requests', {}),
+                dynamic_batching=data.get('dynamic_batching', {}),
                 runtime_args=runtime_args,
             )
         cls._init_from_yaml = False
@@ -113,6 +115,9 @@ def dump(self, data: 'BaseExecutor') -> Dict:
         if hasattr(data, 'requests'):
             r['requests'] = {k: v.__name__ for k, v in data.requests.items()}
 
+        if hasattr(data, 'dynamic_batching'):
+            r['dynamic_batching'] = data.dynamic_batching
+
         if hasattr(data, 'components'):
             r['components'] = data.components
         return r
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index 90a8bb99b363e..36d1b38861507 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -873,6 +873,7 @@ def add(
         uses_after_address: Optional[str] = None,
         uses_before: Optional[Union[str, Type['BaseExecutor'], dict]] = None,
         uses_before_address: Optional[str] = None,
+        uses_dynamic_batching: Optional[dict] = None,
         uses_metas: Optional[dict] = None,
         uses_requests: Optional[dict] = None,
         uses_with: Optional[dict] = None,
@@ -973,6 +974,7 @@ def add(
         :param uses_after_address: The address of the uses-before runtime
         :param uses_before: The executor attached before the Pods described by --uses, typically before sending to all shards, accepted type follows `--uses`. This argument only applies for sharded Deployments (shards > 1).
         :param uses_before_address: The address of the uses-before runtime
+        :param uses_dynamic_batching: Dictionary of keyword arguments that will override the `dynamic_batching` configuration in `uses`
         :param uses_metas: Dictionary of keyword arguments that will override the `metas` configuration in `uses`
         :param uses_requests: Dictionary of keyword arguments that will override the `requests` configuration in `uses`
         :param uses_with: Dictionary of keyword arguments that will override the `with` configuration in `uses`
@@ -1118,6 +1120,7 @@ def add(
         :param uses_after_address: The address of the uses-before runtime
         :param uses_before: The executor attached before the Pods described by --uses, typically before sending to all shards, accepted type follows `--uses`. This argument only applies for sharded Deployments (shards > 1).
         :param uses_before_address: The address of the uses-before runtime
+        :param uses_dynamic_batching: Dictionary of keyword arguments that will override the `dynamic_batching` configuration in `uses`
         :param uses_metas: Dictionary of keyword arguments that will override the `metas` configuration in `uses`
         :param uses_requests: Dictionary of keyword arguments that will override the `requests` configuration in `uses`
         :param uses_with: Dictionary of keyword arguments that will override the `with` configuration in `uses`
diff --git a/jina/parsers/orchestrate/runtimes/worker.py b/jina/parsers/orchestrate/runtimes/worker.py
index 872afa47e0364..b8745cd645085 100644
--- a/jina/parsers/orchestrate/runtimes/worker.py
+++ b/jina/parsers/orchestrate/runtimes/worker.py
@@ -57,6 +57,15 @@ def mixin_worker_runtime_parser(parser):
         Dictionary of keyword arguments that will override the `requests` configuration in `uses`
         ''',
     )
+    gp.add_argument(
+        '--uses-dynamic-batching',
+        action=KVAppendAction,
+        metavar='KEY: VALUE',
+        nargs='*',
+        help='''
+        Dictionary of keyword arguments that will override the `dynamic_batching` configuration in `uses`
+        ''',
+    )
     gp.add_argument(
         '--py-modules',
         type=str,
diff --git a/jina/serve/executors/__init__.py b/jina/serve/executors/__init__.py
index a8d5128df9a84..a22c6591ce41a 100644
--- a/jina/serve/executors/__init__.py
+++ b/jina/serve/executors/__init__.py
@@ -127,6 +127,7 @@ def __init__(
             requests: Optional[Dict] = None,
             runtime_args: Optional[Dict] = None,
             workspace: Optional[str] = None,
+            dynamic_batching: Optional[Dict] = None,
             **kwargs,
     ):
         """`metas` and `requests` are always auto-filled with values from YAML config.
@@ -136,9 +137,11 @@ def __init__(
         :param runtime_args: a dict of arguments injected from :class:`Runtime` during runtime
         :param kwargs: additional extra keyword arguments to avoid failing when extra params ara passed that are not expected
         :param workspace: the workspace of the executor. Only used if a workspace is not already provided in `metas` or `runtime_args`
+        :param dynamic_batching: a dict of endpoint-dynamic_batching config mapping
         """
         self._add_metas(metas)
         self._add_requests(requests)
+        self._add_dynamic_batching(dynamic_batching)
         self._add_runtime_args(runtime_args)
         self._init_instrumentation(runtime_args)
         self._init_monitoring()
@@ -260,6 +263,11 @@ def _add_requests(self, _requests: Optional[Dict]):
                         f'expect {typename(self)}.{func} to be a function, but receiving {typename(_func)}'
                     )
 
+    def _add_dynamic_batching(self, _dynamic_batching: Optional[Dict]):
+        if _dynamic_batching:
+            self.dynamic_batching = getattr(self, 'dynamic_batching', {})
+            self.dynamic_batching.update(_dynamic_batching)
+
     def _add_metas(self, _metas: Optional[Dict]):
         from jina.serve.executors.metas import get_default_metas
 
@@ -450,6 +458,7 @@ def from_hub(
             uses_with: Optional[Dict] = None,
             uses_metas: Optional[Dict] = None,
             uses_requests: Optional[Dict] = None,
+            uses_dynamic_batching: Optional[Dict] = None,
             **kwargs,
     ) -> T:
         """Construct an Executor from Hub.
@@ -459,6 +468,7 @@ def from_hub(
         :param uses_with: dictionary of parameters to overwrite from the default config's with field
         :param uses_metas: dictionary of parameters to overwrite from the default config's metas field
         :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
+        :param uses_dynamic_batching: dictionary of parameters to overwrite from the default config's dynamic_batching field
         :param kwargs: other kwargs accepted by the CLI ``jina hub pull``
         :return: the Hub Executor object.
 
@@ -498,6 +508,7 @@ def from_hub(
             uses_with=uses_with,
             uses_metas=uses_metas,
             uses_requests=uses_requests,
+            uses_dynamic_batching=uses_dynamic_batching,
         )
 
     @classmethod
@@ -507,6 +518,7 @@ def serve(
             uses_metas: Optional[Dict] = None,
             uses_requests: Optional[Dict] = None,
             stop_event: Optional[Union[threading.Event, multiprocessing.Event]] = None,
+            uses_dynamic_batching: Optional[Dict] = None,
             **kwargs,
     ):
         """Serve this Executor in a temporary Flow. Useful in testing an Executor in remote settings.
@@ -516,6 +528,7 @@ def serve(
         :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
         :param stop_event: a threading event or a multiprocessing event that once set will resume the control Flow
             to main thread.
+        :param uses_dynamic_batching: dictionary of parameters to overwrite from the default config's dynamic_batching field
         :param kwargs: other kwargs accepted by the Flow, full list can be found `here <https://docs.jina.ai/api/jina.orchestrate.flow.base/>`
 
         """
@@ -526,6 +539,7 @@ def serve(
             uses_with=uses_with,
             uses_metas=uses_metas,
             uses_requests=uses_requests,
+            uses_dynamic_batching=uses_dynamic_batching,
         )
         with f:
             f.block(stop_event)
@@ -549,6 +563,7 @@ def to_kubernetes_yaml(
             uses_with: Optional[Dict] = None,
             uses_metas: Optional[Dict] = None,
             uses_requests: Optional[Dict] = None,
+            uses_dynamic_batching: Optional[Dict] = None,
             **kwargs,
     ):
         """
@@ -564,6 +579,7 @@ def to_kubernetes_yaml(
         :param uses_with: dictionary of parameters to overwrite from the default config's with field
         :param uses_metas: dictionary of parameters to overwrite from the default config's metas field
         :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
+        :param uses_dynamic_batching: dictionary of parameters to overwrite from the default config's dynamic_batching field
         :param kwargs: other kwargs accepted by the Flow, full list can be found `here <https://docs.jina.ai/api/jina.orchestrate.flow.base/>`
         """
         from jina import Flow
@@ -573,6 +589,7 @@ def to_kubernetes_yaml(
             uses_with=uses_with,
             uses_metas=uses_metas,
             uses_requests=uses_requests,
+            uses_dynamic_batching=uses_dynamic_batching,
         ).to_kubernetes_yaml(
             output_base_path=output_base_path,
             k8s_namespace=k8s_namespace,
@@ -593,6 +610,7 @@ def to_docker_compose_yaml(
             uses_with: Optional[Dict] = None,
             uses_metas: Optional[Dict] = None,
             uses_requests: Optional[Dict] = None,
+            uses_dynamic_batching: Optional[Dict] = None,
             **kwargs,
     ):
         """
@@ -604,6 +622,7 @@ def to_docker_compose_yaml(
         :param uses_with: dictionary of parameters to overwrite from the default config's with field
         :param uses_metas: dictionary of parameters to overwrite from the default config's metas field
         :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
+        :param uses_dynamic_batching: dictionary of parameters to overwrite from the default config's requests field
         :param kwargs: other kwargs accepted by the Flow, full list can be found `here <https://docs.jina.ai/api/jina.orchestrate.flow.base/>`
         """
         from jina import Flow
@@ -613,6 +632,7 @@ def to_docker_compose_yaml(
             uses_with=uses_with,
             uses_metas=uses_metas,
             uses_requests=uses_requests,
+            uses_dynamic_batching=uses_dynamic_batching,
         )
         f.to_docker_compose_yaml(
             output_path=output_path,
diff --git a/jina/serve/executors/decorators.py b/jina/serve/executors/decorators.py
index 46b4da807116b..8ebca2f7f58fe 100644
--- a/jina/serve/executors/decorators.py
+++ b/jina/serve/executors/decorators.py
@@ -44,9 +44,9 @@ def arg_wrapper(self, *args, **kwargs):
 
             if self.__class__ == cls:
                 with ImportExtensions(
-                        required=False,
-                        help_text=f'FileLock is needed to guarantee non-concurrent initialization of replicas in the '
-                                  f'same machine.',
+                    required=False,
+                    help_text=f'FileLock is needed to guarantee non-concurrent initialization of replicas in the '
+                    f'same machine.',
                 ):
                     import filelock
 
@@ -93,7 +93,8 @@ def _inherit_from_parent_class_inner(cls_):
 
 
 def requests(
-        func: Optional[Callable[
+    func: Optional[
+        Callable[
             [
                 'DocumentArray',
                 Dict,
@@ -102,9 +103,10 @@ def requests(
                 List['DocumentArray'],
             ],
             Optional[Union['DocumentArray', Dict]],
-        ]] = None,
-        *,
-        on: Optional[Union[str, Sequence[str]]] = None,
+        ]
+    ] = None,
+    *,
+    on: Optional[Union[str, Sequence[str]]] = None,
 ):
     """
     `@requests` defines the endpoints of an Executor. It has a keyword `on=` to define the endpoint.
@@ -156,9 +158,11 @@ def foo(self, docs, **kwargs):
 
     class FunctionMapper:
         def __init__(self, fn):
+            self._batching_decorator = None
+            fn = self._unwrap_batching_decorator(fn)
             arg_spec = inspect.getfullargspec(fn)
             if not arg_spec.varkw and not __args_executor_func__.issubset(
-                    arg_spec.args
+                arg_spec.args
             ):
                 raise TypeError(
                     f'{fn} accepts only {arg_spec.args} which is fewer than expected, '
@@ -166,9 +170,10 @@ def __init__(self, fn):
                 )
 
             if iscoroutinefunction(fn):
+
                 @functools.wraps(fn)
                 async def arg_wrapper(
-                        executor_instance, *args, **kwargs
+                    executor_instance, *args, **kwargs
                 ):  # we need to get the summary from the executor, so we need to access the self
                     return await fn(executor_instance, *args, **kwargs)
 
@@ -177,33 +182,149 @@ async def arg_wrapper(
 
                 @functools.wraps(fn)
                 def arg_wrapper(
-                        executor_instance, *args, **kwargs
+                    executor_instance, *args, **kwargs
                 ):  # we need to get the summary from the executor, so we need to access the self
                     return fn(executor_instance, *args, **kwargs)
 
                 self.fn = arg_wrapper
 
-        def __set_name__(self, owner, name):
-            _init_requests_by_class(owner)
-            self.fn.class_name = owner.__name__
+        def _unwrap_batching_decorator(self, fn):
+            if type(fn).__name__ == 'DynamicBatchingDecorator':
+                self._batching_decorator = fn
+                return fn.fn
+            else:
+                return fn
+
+        def _inject_owner_attrs(self, owner, name):
+            if not hasattr(owner, 'requests'):
+                owner.requests = {}
             if isinstance(on, (list, tuple)):
                 for o in on:
                     owner.requests_by_class[owner.__name__][o] = self.fn
             else:
-                owner.requests_by_class[owner.__name__][on or __default_endpoint__] = self.fn
-
+                owner.requests_by_class[owner.__name__][
+                    on or __default_endpoint__
+                ] = self.fn
             setattr(owner, name, self.fn)
 
+        def __set_name__(self, owner, name):
+            _init_requests_by_class(owner)
+            if self._batching_decorator:
+                self._batching_decorator._inject_owner_attrs(owner, name)
+            self.fn.class_name = owner.__name__
+            self._inject_owner_attrs(owner, name)
+
+        def __call__(self, *args, **kwargs):
+            # this is needed to make this decorator work in combination with `@requests`
+            return self.fn(*args, **kwargs)
+
     if func:
         return FunctionMapper(func)
     else:
         return FunctionMapper
 
 
+def dynamic_batching(
+    func: Callable[
+        [
+            'DocumentArray',
+            Dict,
+            'DocumentArray',
+            List['DocumentArray'],
+            List['DocumentArray'],
+        ],
+        Optional[Union['DocumentArray', Dict]],
+    ] = None,
+    *,
+    preferred_batch_size: Optional[int] = None,
+    max_batch_size: Optional[int] = None,
+    timeout: Optional[float] = 10_000,
+):
+    """
+    `@dynamic_batching` defines the dynamic batching behavior of an Executor.
+    Dynamic batching works by collecting Documents from multiple requests in a queue, and passing them to the Executor
+    in batches of specified size.
+    This can improve throughput and resource utilization at the cost of increased latency.
+    TODO(johannes) add docstring example
+
+    :param func: the method to decorate
+    :param preferred_batch_size: target number of Documents in a batch. The batcher will collect requests until `preferred_batch_size` is reached,
+        or until `timeout` is reached, or until `max_batch_size` is reached. Therefore, the actual batch size can be smaller or larger than `preferred_batch_size`.
+    :param max_batch_size: maximum number of Documents in a batch. The batcher will not create batches that are larger than `max_batch_size`.
+        However, if a single request contains more than `max_batch_size` Documents, all of these Documents will be passed to the Executor.
+    :param timeout: maximum time in milliseconds to wait for a request to be assigned to a batch.
+        If the oldest request in the queue reaches a waiting time of `timeout`, the batch will be passed to the Executor,
+        even if it contains fewer than `preferred_batch_size` Documents.
+        Default is 10_000ms (10 seconds).
+    :return: decorated function
+    """
+
+    class DynamicBatchingDecorator:
+        def __init__(self, fn):
+            self._requests_decorator = None
+            fn = self._unwrap_requests_decorator(fn)
+            if iscoroutinefunction(fn):
+
+                @functools.wraps(fn)
+                async def arg_wrapper(
+                    executor_instance, *args, **kwargs
+                ):  # we need to get the summary from the executor, so we need to access the self
+                    return await fn(executor_instance, *args, **kwargs)
+
+                self.fn = arg_wrapper
+            else:
+
+                @functools.wraps(fn)
+                def arg_wrapper(
+                    executor_instance, *args, **kwargs
+                ):  # we need to get the summary from the executor, so we need to access the self
+                    return fn(executor_instance, *args, **kwargs)
+
+                self.fn = arg_wrapper
+
+        def _unwrap_requests_decorator(self, fn):
+            if type(fn).__name__ == 'FunctionMapper':
+                self._requests_decorator = fn
+                return fn.fn
+            else:
+                return fn
+
+        def _inject_owner_attrs(self, owner, name):
+            if not hasattr(owner, 'dynamic_batching'):
+                owner.dynamic_batching = {}
+
+            fn_name = self.fn.__name__
+            if not owner.dynamic_batching.get(fn_name):
+                owner.dynamic_batching[fn_name] = {}
+
+            owner.dynamic_batching[fn_name][
+                'preferred_batch_size'
+            ] = preferred_batch_size
+            owner.dynamic_batching[fn_name]['max_batch_size'] = max_batch_size
+            owner.dynamic_batching[fn_name]['timeout'] = timeout
+            setattr(owner, name, self.fn)
+
+        def __set_name__(self, owner, name):
+            _init_requests_by_class(owner)
+            if self._requests_decorator:
+                self._requests_decorator._inject_owner_attrs(owner, name)
+            self.fn.class_name = owner.__name__
+            self._inject_owner_attrs(owner, name)
+
+        def __call__(self, *args, **kwargs):
+            # this is needed to make this decorator work in combination with `@requests`
+            return self.fn(*args, **kwargs)
+
+    if func:
+        return DynamicBatchingDecorator(func)
+    else:
+        return DynamicBatchingDecorator
+
+
 def monitor(
-        *,
-        name: Optional[str] = None,
-        documentation: Optional[str] = None,
+    *,
+    name: Optional[str] = None,
+    documentation: Optional[str] = None,
 ):
     """
     Decorator and context manager that allows monitoring of an Executor.
diff --git a/jina/serve/executors/metas.py b/jina/serve/executors/metas.py
index 6617d48fe37ed..d477cee8d5588 100644
--- a/jina/serve/executors/metas.py
+++ b/jina/serve/executors/metas.py
@@ -24,7 +24,7 @@ def get_executor_taboo():
     Returns a set of executor meta variables
     :return: set of executor meta variables
     """
-    taboo = {'self', 'args', 'kwargs', 'metas', 'requests', 'runtime_args'}
+    taboo = {'self', 'args', 'kwargs', 'metas', 'requests', 'runtime_args', 'dynamic_batching'}
     _defaults = get_default_metas()
     taboo.update(_defaults.keys())
     return taboo
diff --git a/jina/serve/runtimes/worker/batch_queue.py b/jina/serve/runtimes/worker/batch_queue.py
new file mode 100644
index 0000000000000..e92dc2ff32de3
--- /dev/null
+++ b/jina/serve/runtimes/worker/batch_queue.py
@@ -0,0 +1,172 @@
+import asyncio
+from asyncio import Event, Task
+from typing import Any, Dict, List, Optional
+
+from jina import DocumentArray
+from jina.serve.executors import BaseExecutor
+from jina.types.request.data import DataRequest
+
+
+async def sleep_then_set(time_seconds: float, event: Event):
+    """Sleep for time_seconds and then set the event
+
+    :param time_seconds: time to sleep
+    :param event: event to set
+    """
+    await asyncio.sleep(time_seconds)
+    event.set()
+
+
+class BatchQueue:
+    """A batch queue that holds the data request and the executor."""
+
+    def __init__(
+        self,
+        executor: BaseExecutor,
+        exec_endpoint: str,
+        args: Any,
+        params: Optional[Dict] = None,
+        preferred_batch_size: int = 4,
+        timeout: int = 10_000,
+        max_batch_size: Optional[int] = None,
+    ) -> None:
+        if params is None:
+            params = dict()
+        self._executor: BaseExecutor = executor
+        self._exec_endpoint: str = exec_endpoint
+        self.args = args
+        self.params = params
+
+        self._preferred_batch_size: int = preferred_batch_size
+        self._timeout: int = timeout
+        self._max_batch_size: Optional[int] = max_batch_size
+        if self._max_batch_size:
+            if self._max_batch_size < self._preferred_batch_size:
+                raise ValueError(
+                    f'Max batch size must be greater than or equal to preferred batch size.'
+                    f'Got max batch size {self._max_batch_size} and preferred'
+                    f'batch size {self._preferred_batch_size}'
+                )
+
+        self._reset()
+
+    def __repr__(self) -> str:
+        return (
+            f'{self.__class__.__name__}({self._executor}, {self._exec_endpoint},'
+            f'preferred_batch_size={self._preferred_batch_size}, timeout={self._timeout})'
+        )
+
+    def __str__(self) -> str:
+        return self.__repr__()
+
+    def _reset(self) -> None:
+        """Set all events and reset the batch queue."""
+        self._requests: List[DataRequest] = []
+        self._request_lens: List[int] = []
+        self._big_doc: DocumentArray = DocumentArray.empty()
+
+        self._flush_trigger: Event = Event()
+        self._flush_task: Task = asyncio.create_task(
+            self.await_then_flush(self._flush_trigger)
+        )
+        self._timer_started: bool = False
+
+    def _start_timer(self):
+        asyncio.create_task(sleep_then_set(self._timeout / 1000, self._flush_trigger))
+        self._timer_started = True
+
+    async def push(self, request: DataRequest) -> Task:
+        """Append request to the queue. Once the request has been processed, the returned task will complete.
+
+        :param request: The request to append to the queue.
+
+        :return: The task that will be set once the request has been processed.
+        """
+        if self._max_batch_size:
+            if len(self._big_doc) + len(request.docs) > self._max_batch_size:
+                self._flush_trigger.set()
+                await self._flush_task
+
+        if not self._timer_started:
+            self._start_timer()
+
+        docs = request.docs
+        self._big_doc.extend(docs)
+        self._requests.append(request)
+        self._request_lens.append(len(docs))
+
+        if len(self._big_doc) >= self._preferred_batch_size:
+            self._flush_trigger.set()
+
+        return self._flush_task
+
+    async def await_then_flush(self, trigger_event: Event) -> None:
+        """Process all requests in the queue once event is set.
+
+        :param trigger_event: The event that will trigger the flush.
+        """
+        await trigger_event.wait()
+        try:
+            # The executor might accidentally change the size
+            input_len_before_call: int = len(self._big_doc)
+
+            if self._max_batch_size:
+                if input_len_before_call > self._max_batch_size:
+                    # TODO: We might want to have the batch queue split the request up instead
+                    # TODO(johannes): update the docstring of the decorator to reflect this
+                    raise ValueError(
+                        f'Input length {input_len_before_call} is greater than max batch size {self._max_batch_size}'
+                    )
+
+            # We need to get the executor to process the big doc
+            return_data = await self._executor.__acall__(
+                req_endpoint=self._exec_endpoint,
+                docs=self._big_doc,
+                parameters=self.params,
+                docs_matrix=None,  # joining manually with batch queue is not supported right now
+                # TO GIRISH: The tracing seems to work already.
+                # Do we need to pass this?
+                # The only difference I can see is that the nesting level is different.
+                tracing_context=None,
+            )
+
+            # Output validation
+            if isinstance(return_data, DocumentArray):
+                assert (
+                    len(return_data) == input_len_before_call
+                ), f'Dynamic Batching requires input size to equal output size. Expected output size {input_len_before_call}, but got {len(return_data)}'
+            elif return_data is None:
+                assert (
+                    len(self._big_doc) == input_len_before_call
+                ), f'Dynamic Batching requires input size to equal output size. Expected output size {input_len_before_call}, but got {len(self._big_doc)}'
+            else:
+                raise TypeError(
+                    f'The return type must be DocumentArray / `None` when using dynamic batching, '
+                    f'but getting {return_data!r}'
+                )
+
+            # We need to reslice the big doc array into the original requests
+            self._fan_out_return_data(return_data)
+
+            self._reset()
+
+        except Exception as e:
+            # We need to reset the batch queue
+            # This needs to occur even if the executor fails otherwise it will block forever
+            self._reset()
+            raise e
+
+    def _fan_out_return_data(self, return_data: Optional[DocumentArray]):
+        consumed_count: int = 0
+        for request, request_len in zip(self._requests, self._request_lens):
+            left = consumed_count
+            right = consumed_count + request_len
+            if return_data:
+                request.data.set_docs_convert_arrays(
+                    return_data[left:right], ndarray_type=self.args.output_array_type
+                )
+            else:
+                request.data.set_docs_convert_arrays(
+                    self._big_doc[left:right], ndarray_type=self.args.output_array_type
+                )
+            consumed_count += request_len
diff --git a/jina/serve/runtimes/worker/request_handling.py b/jina/serve/runtimes/worker/request_handling.py
index ad1a2bfde95b3..9bfe6594cc627 100644
--- a/jina/serve/runtimes/worker/request_handling.py
+++ b/jina/serve/runtimes/worker/request_handling.py
@@ -6,6 +6,7 @@
 from jina.excepts import BadConfigSource
 from jina.importer import ImportExtensions
 from jina.serve.executors import BaseExecutor
+from jina.serve.runtimes.worker.batch_queue import BatchQueue
 from jina.types.request.data import DataRequest
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -59,6 +60,53 @@ def __init__(
         )
         self._init_monitoring(metrics_registry, meter)
         self.deployment_name = deployment_name
+        # In order to support batching parameters separately, we have to lazily create batch queues
+        # So we store the config for each endpoint in the initialization
+        self._batchqueue_config: Dict[str, Dict] = {}
+        # the below is of "shape" exec_endpoint_name -> parameters_key -> batch_queue
+        self._batchqueue_instances: Dict[
+            str, Dict[str, BatchQueue]
+        ] = self._init_batchqueue_dict()
+
+    def _init_batchqueue_dict(self) -> Dict[str, Dict[str, BatchQueue]]:
+        """Determines how endpoints and method names map to batch queues, without instantiating them.
+        :return: a dictionary of "shape" exec_endpoint_name -> parameters_key -> batch_queue.
+            The `exec_endpoint_name` keys are filled in, the rest is empty.
+        """
+        if getattr(self._executor, 'dynamic_batching', None) is not None:
+            # We need to sort the keys into endpoints and functions
+            # Endpoints allow specific configurations while functions allow configs to be applied to all endpoints of the function
+            dbatch_endpoints = []
+            dbatch_functions = []
+            for key, dbatch_config in self._executor.dynamic_batching.items():
+                if key.startswith('/'):
+                    dbatch_endpoints.append((key, dbatch_config))
+                else:
+                    dbatch_functions.append((key, dbatch_config))
+
+            # Specific endpoint configs take precedence over function configs
+            for endpoint, dbatch_config in dbatch_endpoints:
+                self._batchqueue_config[endpoint] = dbatch_config
+
+            # Process function configs
+            func_endpoints: Dict[str, List[str]] = {
+                func.__name__: [] for func in self._executor.requests.values()
+            }
+            for endpoint, func in self._executor.requests.items():
+                func_endpoints[func.__name__].append(endpoint)
+            for func_name, dbatch_config in dbatch_functions:
+                for endpoint in func_endpoints[func_name]:
+                    if endpoint not in self._batchqueue_config:
+                        self._batchqueue_config[endpoint] = dbatch_config
+
+            self.logger.debug(
+                f'Executor Dynamic Batching configs: {self._executor.dynamic_batching}'
+            )
+            self.logger.debug(
+                f'Endpoint Batch Queue Configs: {self._batchqueue_config}'
+            )
+
+            return {endpoint: {} for endpoint in self._batchqueue_config.keys()}
 
     def _init_monitoring(
         self,
@@ -143,6 +191,7 @@ def _load_executor(
                 uses_with=self.args.uses_with,
                 uses_metas=self.args.uses_metas,
                 uses_requests=self.args.uses_requests,
+                uses_dynamic_batching=self.args.uses_dynamic_batching,
                 runtime_args={  # these are not parsed to the yaml config file but are pass directly during init
                     'workspace': self.args.workspace,
                     'shard_id': self.args.shard_id,
@@ -169,21 +218,28 @@ def _load_executor(
             raise
 
     def _refresh_executor(self, changed_files):
+        import copy
         import importlib
         import inspect
-        import copy
         import sys
 
         try:
-            sys_mod_files_modules = {getattr(module, '__file__', ''): module for module in sys.modules.values()}
+            sys_mod_files_modules = {
+                getattr(module, '__file__', ''): module
+                for module in sys.modules.values()
+            }
 
             for file in changed_files:
                 if file in sys_mod_files_modules:
                     importlib.reload(sys_mod_files_modules[file])
                 else:
-                    self.logger.debug(f'Changed file {file} was not previously imported.')
+                    self.logger.debug(
+                        f'Changed file {file} was not previously imported.'
+                    )
         except Exception as exc:
-            self.logger.error(f' Exception when refreshing Executor when changes detected in {changed_files}')
+            self.logger.error(
+                f' Exception when refreshing Executor when changes detected in {changed_files}'
+            )
             raise exc
 
         importlib.reload(inspect.getmodule(self._executor.__class__))
@@ -201,7 +257,7 @@ def _refresh_executor(self, changed_files):
 
     @staticmethod
     def _parse_params(parameters: Dict, executor_name: str):
-        parsed_params = parameters
+        parsed_params = parameters  # TODO: Copy? This still points to the same object
         specific_parameters = parameters.get(executor_name, None)
         if specific_parameters:
             parsed_params.update(**specific_parameters)
@@ -299,43 +355,63 @@ async def handle(
         :param tracing_context: Optional OpenTelemetry tracing context from the originating request.
         :returns: the processed message
         """
+
         # skip executor if endpoints mismatch
-        if (
-            requests[0].header.exec_endpoint not in self._executor.requests
-            and __default_endpoint__ not in self._executor.requests
-        ):
-            self.logger.debug(
-                f'skip executor: mismatch request, exec_endpoint: {requests[0].header.exec_endpoint}, requests: {self._executor.requests}'
-            )
-            return requests[0]
+        exec_endpoint: str = requests[0].header.exec_endpoint
+        if exec_endpoint not in self._executor.requests:
+            if __default_endpoint__ in self._executor.requests:
+                exec_endpoint = __default_endpoint__
+            else:
+                self.logger.debug(
+                    f'skip executor: mismatch request, exec_endpoint: {exec_endpoint}, requests: {self._executor.requests}'
+                )
+                return requests[0]
 
         self._record_request_size_monitoring(requests)
 
         params = self._parse_params(requests[0].parameters, self._executor.metas.name)
-        docs = WorkerRequestHandler.get_docs_from_request(
-            requests,
-            field='docs',
-        )
 
-        # executor logic
-        docs_matrix, docs_map = WorkerRequestHandler._get_docs_matrix_from_request(
-            requests
-        )
-        return_data = await self._executor.__acall__(
-            req_endpoint=requests[0].header.exec_endpoint,
-            docs=docs,
-            parameters=params,
-            docs_matrix=docs_matrix,
-            docs_map=docs_map,
-            tracing_context=tracing_context,
-        )
+        if exec_endpoint in self._batchqueue_config:
+            assert len(requests) == 1, 'dynamic batching does not support no_reduce'
+
+            param_key = str(params)
+            if param_key not in self._batchqueue_instances[exec_endpoint]:
+                self._batchqueue_instances[exec_endpoint][param_key] = BatchQueue(
+                    executor=self._executor,
+                    exec_endpoint=exec_endpoint,
+                    args=self.args,
+                    params=params,
+                    **self._batchqueue_config[exec_endpoint],
+                )
+            # This is necessary because push might need to await for the queue to be emptied
+            task = await self._batchqueue_instances[exec_endpoint][param_key].push(
+                requests[0]
+            )
+            await task
+        else:
+            docs = WorkerRequestHandler.get_docs_from_request(
+                requests,
+                field='docs',
+            )
+
+            docs_matrix, docs_map = WorkerRequestHandler._get_docs_matrix_from_request(
+                requests
+            )
+            return_data = await self._executor.__acall__(
+                req_endpoint=requests[0].header.exec_endpoint,
+                docs=docs,
+                parameters=params,
+                docs_matrix=docs_matrix,
+                docs_map=docs_map,
+                tracing_context=tracing_context,
+            )
 
-        docs = self._set_result(requests, return_data, docs)
+            docs = self._set_result(requests, return_data, docs)
 
         for req in requests:
             req.add_executor(self.deployment_name)
 
-        self._record_docs_processed_monitoring(requests, docs)
+        self._record_docs_processed_monitoring(requests, requests[0].docs)
         self._record_response_size_monitoring(requests)
 
         return requests[0]
diff --git a/jina_cli/autocomplete.py b/jina_cli/autocomplete.py
index 2d646df0c24a9..e8b41c4e9f0ba 100644
--- a/jina_cli/autocomplete.py
+++ b/jina_cli/autocomplete.py
@@ -37,6 +37,7 @@
             '--uses-with',
             '--uses-metas',
             '--uses-requests',
+            '--uses-dynamic-batching',
             '--py-modules',
             '--output-array-type',
             '--exit-on-exceptions',
@@ -252,6 +253,7 @@
             '--uses-with',
             '--uses-metas',
             '--uses-requests',
+            '--uses-dynamic-batching',
             '--py-modules',
             '--output-array-type',
             '--exit-on-exceptions',
@@ -312,6 +314,7 @@
             '--uses-with',
             '--uses-metas',
             '--uses-requests',
+            '--uses-dynamic-batching',
             '--py-modules',
             '--output-array-type',
             '--exit-on-exceptions',
diff --git a/tests/unit/serve/dynamic_batching/__init__.py b/tests/unit/serve/dynamic_batching/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/unit/serve/dynamic_batching/test_batch_queue.py b/tests/unit/serve/dynamic_batching/test_batch_queue.py
new file mode 100644
index 0000000000000..3f5f780de7fa2
--- /dev/null
+++ b/tests/unit/serve/dynamic_batching/test_batch_queue.py
@@ -0,0 +1,254 @@
+import asyncio
+import time
+
+import pytest
+
+from jina import Document, DocumentArray, Executor, requests
+from jina.serve.runtimes.worker.batch_queue import BatchQueue, sleep_then_set
+from jina.types.request.data import DataRequest
+
+
+@pytest.mark.skip(reason="Sleep then set doesnt count normal sleep here")
+@pytest.mark.asyncio
+async def test_sleep_then_set():
+    event = asyncio.Event()
+    asyncio.create_task(sleep_then_set(0.1, event))
+    assert not event.is_set()
+    await asyncio.sleep(0)
+    assert not event.is_set()
+    await asyncio.sleep(0.2)
+    assert event.is_set()
+
+    event = asyncio.Event()
+    asyncio.create_task(sleep_then_set(0.1, event))
+    assert not event.is_set()
+    await asyncio.sleep(0)
+    assert not event.is_set()
+    time.sleep(0.2)
+    await asyncio.sleep(0)
+    assert event.is_set()
+
+    event = asyncio.Event()
+    asyncio.create_task(sleep_then_set(0.1, event))
+    for i in range(3):
+        assert not event.is_set()
+        time.sleep(0.1)
+    assert not event.is_set()
+    await asyncio.sleep(0)
+    assert event.is_set()
+
+
+@pytest.mark.asyncio
+async def test_batch_queue():
+    class MockExecutor(Executor):
+        @requests
+        def foo(self, docs, **kwargs):
+            return DocumentArray([Document(text='Done') for _ in docs])
+
+    executor = MockExecutor()
+    args = executor.runtime_args
+    args.output_array_type = None
+    bq: BatchQueue = BatchQueue(
+        executor=executor,
+        exec_endpoint="/",
+        args=args,
+        preferred_batch_size=4,
+        timeout=500,
+    )
+
+    data_requests = [DataRequest() for _ in range(4)]
+    for req in data_requests:
+        req.data.docs = DocumentArray.empty(1)
+        assert req.data.docs[0].text == ''
+
+    # Test preferred batch size
+    tasks = [await bq.push(req) for req in data_requests[:-1]]
+    assert len(bq._requests) == 3
+    assert all(not task.done() for task in tasks)
+
+    tasks.append(await bq.push(data_requests[-1]))
+    assert len(bq._requests) == 4
+    assert all(not task.done() for task in tasks)
+
+    # The time here is necessary because the flush has awaits inside
+    # We need to give time for the flush to finish
+    await asyncio.sleep(0.2)
+    assert len(bq._requests) == 0
+    assert len(bq._big_doc) == 0
+    assert all(task.done() for task in tasks)
+    assert all(task.exception() is None for task in tasks)
+
+    for req in data_requests:
+        assert req.data.docs[0].text == 'Done'
+
+    # Test timeout
+    for req in data_requests:
+        req.data.docs[0].text = 'Not Done'
+
+    single_task = await bq.push(data_requests[0])
+    assert not single_task.done()
+    assert data_requests[0].data.docs[0].text == 'Not Done'
+
+    await asyncio.sleep(0)
+    assert not single_task.done()
+    assert data_requests[0].data.docs[0].text == 'Not Done'
+
+    await asyncio.sleep(0.1)
+    assert not single_task.done()
+    assert data_requests[0].data.docs[0].text == 'Not Done'
+
+    await asyncio.sleep(0.5)
+    assert single_task.done()
+    assert data_requests[0].data.docs[0].text == 'Done'
+
+
+@pytest.mark.asyncio
+async def test_batch_queue_max_batch_size():
+    class MockExecutor(Executor):
+        @requests
+        def foo(self, docs, **kwargs):
+            return DocumentArray([Document(text='Done') for _ in docs])
+
+    executor = MockExecutor()
+    args = executor.runtime_args
+    args.output_array_type = None
+
+    bq: BatchQueue = BatchQueue(
+        executor=executor,
+        exec_endpoint="/",
+        args=args,
+        preferred_batch_size=4,
+        max_batch_size=5,
+        timeout=600,
+    )
+
+    data_requests = [DataRequest() for _ in range(3)]
+    for req in data_requests:
+        req.data.docs = DocumentArray.empty(1)
+        assert req.data.docs[0].text == ''
+
+    # Test preferred batch size
+    tasks = [await bq.push(req) for req in data_requests]
+    assert len(bq._requests) == 3
+    assert len(bq._big_doc) == 3
+    assert all(not task.done() for task in tasks)
+    await asyncio.sleep(0.2)
+    assert len(bq._requests) == 3
+    assert len(bq._big_doc) == 3
+    assert all(not task.done() for task in tasks)
+
+    data_request_b = DataRequest()
+    data_request_b.data.docs = DocumentArray.empty(3)
+    taskb = await bq.push(data_request_b)
+
+    assert all(task.done() for task in tasks)
+
+    assert len(bq._requests) == 1
+    assert len(bq._big_doc) == 3
+    assert not taskb.done()
+    await asyncio.sleep(0.2)
+    assert not taskb.done()
+
+    data_request_c = DataRequest()
+    data_request_c.data.docs = DocumentArray.empty(2)
+    taskc = await bq.push(data_request_c)
+
+    await asyncio.sleep(0.2)
+    assert len(bq._requests) == 0
+    assert len(bq._big_doc) == 0
+    assert taskb.done()
+    assert taskc.done()
+
+
+@pytest.mark.asyncio
+async def test_batch_queue_max_batch_size_exception():
+    class MockExecutor(Executor):
+        @requests
+        def foo(self, docs, **kwargs):
+            return DocumentArray([Document(text='Done') for _ in docs])
+
+    executor = MockExecutor()
+    args = executor.runtime_args
+    args.output_array_type = None
+
+    with pytest.raises(ValueError):
+        bq: BatchQueue = BatchQueue(
+            executor=executor,
+            exec_endpoint="/",
+            args=args,
+            preferred_batch_size=4,
+            max_batch_size=2,
+            timeout=500,
+        )
+
+    bq: BatchQueue = BatchQueue(
+        executor=executor,
+        exec_endpoint="/",
+        args=args,
+        preferred_batch_size=4,
+        max_batch_size=5,
+        timeout=600,
+    )
+
+    data_request_b = DataRequest()
+    data_request_b.data.docs = DocumentArray.empty(6)
+    taskb = await bq.push(data_request_b)
+
+    await asyncio.sleep(0.2)
+    assert taskb.done()
+    assert isinstance(taskb.exception(), ValueError)
+
+
+@pytest.mark.asyncio
+async def test_exception():
+    class MockExecutor(Executor):
+        @requests
+        def foo(self, docs, **kwargs):
+            return "Bad type"
+
+    executor = MockExecutor()
+    args = executor.runtime_args
+    args.output_array_type = None
+    bq: BatchQueue = BatchQueue(
+        executor=executor,
+        exec_endpoint="/",
+        args=args,
+        preferred_batch_size=4,
+        timeout=500,
+    )
+
+    data_requests = [DataRequest() for _ in range(4)]
+    for req in data_requests:
+        req.data.docs = DocumentArray.empty(1)
+        assert req.data.docs[0].text == ''
+
+    tasks = [await bq.push(req) for req in data_requests]
+    assert len(bq._requests) == 4
+    assert all(not task.done() for task in tasks)
+
+    for i in range(3):
+        with pytest.raises(TypeError):
+            await tasks[i]
+        assert tasks[i].done()
+        assert tasks[i].exception is not None
+
+
+@pytest.mark.asyncio
+async def test_repr_and_str():
+    class MockExecutor(Executor):
+        @requests
+        def foo(self, docs, **kwargs):
+            return DocumentArray([Document(text='Done') for _ in docs])
+
+    executor = MockExecutor()
+    args = executor.runtime_args
+    args.output_array_type = None
+    bq: BatchQueue = BatchQueue(
+        executor=executor,
+        exec_endpoint="/",
+        args=args,
+        preferred_batch_size=4,
+        timeout=500,
+    )
+
+    assert repr(bq) == str(bq)
diff --git a/tests/unit/serve/dynamic_batching/test_dynamic_batching.py b/tests/unit/serve/dynamic_batching/test_dynamic_batching.py
new file mode 100644
index 0000000000000..238a1c4715739
--- /dev/null
+++ b/tests/unit/serve/dynamic_batching/test_dynamic_batching.py
@@ -0,0 +1,345 @@
+import multiprocessing as mp
+import time
+from collections import namedtuple
+
+import pytest
+
+from jina import (
+    Client,
+    Document,
+    DocumentArray,
+    Executor,
+    Flow,
+    dynamic_batching,
+    requests,
+)
+
+TIMEOUT_TOLERANCE = 1
+FOO_SUCCESS_MSG = 'Done through foo'
+BAR_SUCCESS_MSG = 'Done through bar'
+
+
+class PlaceholderExecutor(Executor):
+    @requests(on=['/foo'])
+    @dynamic_batching(preferred_batch_size=4)
+    def foo_fun(self, docs, **kwargs):
+        for doc in docs:
+            doc.text += FOO_SUCCESS_MSG
+
+    @requests(on=['/bar', '/baz'])
+    @dynamic_batching(preferred_batch_size=4, timeout=2000)
+    def bar_fun(self, docs, **kwargs):
+        for doc in docs:
+            doc.text += BAR_SUCCESS_MSG
+        return docs
+
+    @requests(on=['/wrongtype'])
+    @dynamic_batching(preferred_batch_size=4, timeout=2000)
+    def wrong_return_type_fun(self, docs, **kwargs):
+        return 'Fail me!'
+
+    @requests(on=['/wronglenda'])
+    @dynamic_batching(preferred_batch_size=4, timeout=2000)
+    def wrong_return_lenda_fun(self, docs, **kwargs):
+        return DocumentArray.empty(len(docs) + 1)
+
+    @requests(on=['/wronglennone'])
+    @dynamic_batching(preferred_batch_size=4, timeout=2000)
+    def wrong_return_lennone_fun(self, docs, **kwargs):
+        docs.append(Document())
+
+    @requests(on=['/param'])
+    @dynamic_batching(preferred_batch_size=4, timeout=2000)
+    def param_fun(self, docs, parameters, **kwargs):
+        for doc in docs:
+            doc.text += str(parameters)
+
+
+RequestStruct = namedtuple('RequestStruct', ['port', 'endpoint', 'iterator'])
+
+
+def call_api(req: RequestStruct):
+    c = Client(port=req.port)
+    return c.post(
+        req.endpoint,
+        inputs=DocumentArray([Document(text=str(i)) for i in req.iterator]),
+    )
+
+
+RequestStructParams = namedtuple(
+    'RequestStructParams', ['port', 'endpoint', 'iterator', 'params']
+)
+
+
+def call_api_with_params(req: RequestStructParams):
+    c = Client(port=req.port)
+    return c.post(
+        req.endpoint,
+        inputs=DocumentArray([Document(text=str(i)) for i in req.iterator]),
+        parameters=req.params,
+    )
+
+
+def test_timeout():
+    f = Flow().add(uses=PlaceholderExecutor)
+    with f:
+        start_time = time.time()
+        f.post('/bar', inputs=DocumentArray.empty(2))
+        time_taken = time.time() - start_time
+        assert time_taken > 2, 'Timeout ended too fast'
+        assert time_taken < 2 + TIMEOUT_TOLERANCE, 'Timeout ended too slowly'
+
+        with mp.Pool(3) as p:
+            start_time = time.time()
+            list(
+                p.map(
+                    call_api,
+                    [
+                        RequestStruct(f.port, '/bar', range(1)),
+                        RequestStruct(f.port, '/bar', range(1)),
+                        RequestStruct(f.port, '/bar', range(1)),
+                    ],
+                )
+            )
+            time_taken = time.time() - start_time
+            assert time_taken > 2, 'Timeout ended too fast'
+            assert time_taken < 2 + TIMEOUT_TOLERANCE, 'Timeout ended too slowly'
+
+
+def test_preferred_batch_size():
+    f = Flow().add(uses=PlaceholderExecutor)
+    with f:
+        with mp.Pool(2) as p:
+            start_time = time.time()
+            list(
+                p.map(
+                    call_api,
+                    [
+                        RequestStruct(f.port, '/bar', range(2)),
+                        RequestStruct(f.port, '/bar', range(2)),
+                    ],
+                )
+            )
+            time_taken = time.time() - start_time
+            assert time_taken < TIMEOUT_TOLERANCE
+
+        with mp.Pool(2) as p:
+            start_time = time.time()
+            list(
+                p.map(
+                    call_api,
+                    [
+                        RequestStruct(f.port, '/bar', range(3)),
+                        RequestStruct(f.port, '/bar', range(2)),
+                    ],
+                )
+            )
+            time_taken = time.time() - start_time
+            assert time_taken < TIMEOUT_TOLERANCE
+
+        with mp.Pool(4) as p:
+            start_time = time.time()
+            list(
+                p.map(
+                    call_api,
+                    [
+                        RequestStruct(f.port, '/bar', range(1)),
+                        RequestStruct(f.port, '/bar', range(1)),
+                        RequestStruct(f.port, '/bar', range(1)),
+                        RequestStruct(f.port, '/bar', range(1)),
+                    ],
+                )
+            )
+            time_taken = time.time() - start_time
+            assert time_taken < TIMEOUT_TOLERANCE
+
+
+def test_correctness():
+    f = Flow().add(uses=PlaceholderExecutor)
+    with f:
+        with mp.Pool(2) as p:
+            results = list(
+                p.map(
+                    call_api,
+                    [
+                        RequestStruct(f.port, '/bar', 'ab'),
+                        RequestStruct(f.port, '/bar', 'cd'),
+                    ],
+                )
+            )
+            assert all([len(result) == 2 for result in results])
+            assert [doc.text for doc in results[0]] == [
+                f'a{BAR_SUCCESS_MSG}',
+                f'b{BAR_SUCCESS_MSG}',
+            ]
+            assert [doc.text for doc in results[1]] == [
+                f'c{BAR_SUCCESS_MSG}',
+                f'd{BAR_SUCCESS_MSG}',
+            ]
+
+        with mp.Pool(2) as p:
+            results = list(
+                p.map(
+                    call_api,
+                    [
+                        RequestStruct(f.port, '/foo', 'ABC'),
+                        RequestStruct(f.port, '/foo', 'AB'),
+                    ],
+                )
+            )
+            assert [len(r) for r in results] == [3, 2]
+            assert [doc.text for doc in results[0]] == [
+                f'A{FOO_SUCCESS_MSG}',
+                f'B{FOO_SUCCESS_MSG}',
+                f'C{FOO_SUCCESS_MSG}',
+            ]
+            assert [doc.text for doc in results[1]] == [
+                f'A{FOO_SUCCESS_MSG}',
+                f'B{FOO_SUCCESS_MSG}',
+            ]
+
+        # This is the only one that waits for timeout because its slow
+        # But we should still test it at least once
+        with mp.Pool(3) as p:
+            results = list(
+                p.map(
+                    call_api,
+                    [
+                        RequestStruct(f.port, '/bar', 'a'),
+                        RequestStruct(f.port, '/bar', 'b'),
+                        RequestStruct(f.port, '/bar', 'c'),
+                    ],
+                )
+            )
+            assert all([len(result) == 1 for result in results])
+            assert [doc.text for doc in results[0]] == [f'a{BAR_SUCCESS_MSG}']
+            assert [doc.text for doc in results[1]] == [f'b{BAR_SUCCESS_MSG}']
+            assert [doc.text for doc in results[2]] == [f'c{BAR_SUCCESS_MSG}']
+
+
+def test_param_correctness():
+    f = Flow().add(uses=PlaceholderExecutor)
+    with f:
+        with mp.Pool(2) as p:
+            PARAM = {'key': 'value'}
+            results = list(
+                p.map(
+                    call_api_with_params,
+                    [
+                        RequestStructParams(f.port, '/param', 'ab', PARAM),
+                        RequestStructParams(f.port, '/param', 'cd', PARAM),
+                    ],
+                )
+            )
+            assert all([len(result) == 2 for result in results])
+            assert [doc.text for doc in results[0]] == [
+                f'a{str(PARAM)}',
+                f'b{str(PARAM)}',
+            ]
+            assert [doc.text for doc in results[1]] == [
+                f'c{str(PARAM)}',
+                f'd{str(PARAM)}',
+            ]
+
+        with mp.Pool(2) as p:
+            PARAM1 = {'key1': 'value1'}
+            PARAM2 = {'key2': 'value2'}
+            results = list(
+                p.map(
+                    call_api_with_params,
+                    [
+                        RequestStructParams(f.port, '/param', 'ABCD', PARAM1),
+                        RequestStructParams(f.port, '/param', 'ABCD', PARAM2),
+                    ],
+                )
+            )
+            assert [len(r) for r in results] == [4, 4]
+            assert [doc.text for doc in results[0]] == [
+                f'A{str(PARAM1)}',
+                f'B{str(PARAM1)}',
+                f'C{str(PARAM1)}',
+                f'D{str(PARAM1)}',
+            ]
+            assert [doc.text for doc in results[1]] == [
+                f'A{str(PARAM2)}',
+                f'B{str(PARAM2)}',
+                f'C{str(PARAM2)}',
+                f'D{str(PARAM2)}',
+            ]
+
+        with mp.Pool(3) as p:
+            PARAM1 = {'key1': 'value1'}
+            PARAM2 = {'key2': 'value2'}
+            results = list(
+                p.map(
+                    call_api_with_params,
+                    [
+                        RequestStructParams(f.port, '/param', 'ABC', PARAM1),
+                        RequestStructParams(f.port, '/param', 'ABCD', PARAM2),
+                        RequestStructParams(f.port, '/param', 'D', PARAM1),
+                    ],
+                )
+            )
+            assert [len(r) for r in results] == [3, 4, 1]
+            assert [doc.text for doc in results[0]] == [
+                f'A{str(PARAM1)}',
+                f'B{str(PARAM1)}',
+                f'C{str(PARAM1)}',
+            ]
+            assert [doc.text for doc in results[1]] == [
+                f'A{str(PARAM2)}',
+                f'B{str(PARAM2)}',
+                f'C{str(PARAM2)}',
+                f'D{str(PARAM2)}',
+            ]
+            assert [doc.text for doc in results[2]] == [f'D{str(PARAM1)}']
+
+
+def test_failure_propagation():
+    from jina.excepts import BadServer
+
+    f = Flow().add(uses=PlaceholderExecutor)
+    with f:
+        with pytest.raises(BadServer):
+            Client(port=f.port).post(
+                '/wrongtype',
+                inputs=DocumentArray([Document(text=str(i)) for i in range(4)]),
+            )
+        with pytest.raises(BadServer):
+            Client(port=f.port).post(
+                '/wrongtype',
+                inputs=DocumentArray([Document(text=str(i)) for i in range(2)]),
+            )
+        with pytest.raises(BadServer):
+            Client(port=f.port).post(
+                '/wrongtype',
+                inputs=DocumentArray([Document(text=str(i)) for i in range(8)]),
+            )
+        with pytest.raises(BadServer):
+            Client(port=f.port).post(
+                '/wronglenda',
+                inputs=DocumentArray([Document(text=str(i)) for i in range(8)]),
+            )
+        with pytest.raises(BadServer):
+            Client(port=f.port).post(
+                '/wronglennone',
+                inputs=DocumentArray([Document(text=str(i)) for i in range(8)]),
+            )
+
+
+def test_specific_endpoint_batching():
+    f = Flow().add(
+        uses=PlaceholderExecutor,
+        uses_dynamic_batching={'/baz': {'preferred_batch_size': 2, 'timeout': 1000}},
+    )
+    with f:
+        start_time = time.time()
+        f.post('/bar', inputs=DocumentArray.empty(2))
+        time_taken = time.time() - start_time
+        assert time_taken > 2, 'Timeout ended too fast'
+        assert time_taken < 2 + TIMEOUT_TOLERANCE, 'Timeout ended too slowly'
+
+        start_time = time.time()
+        f.post('/baz', inputs=DocumentArray.empty(2))
+        time_taken = time.time() - start_time
+        assert time_taken < TIMEOUT_TOLERANCE, 'Timeout ended too slowly'
diff --git a/tests/unit/serve/dynamic_batching/test_dynamic_batching_config.py b/tests/unit/serve/dynamic_batching/test_dynamic_batching_config.py
new file mode 100644
index 0000000000000..bbc0739580ded
--- /dev/null
+++ b/tests/unit/serve/dynamic_batching/test_dynamic_batching_config.py
@@ -0,0 +1,31 @@
+from pathlib import Path
+
+from jina import Document, DocumentArray, Executor, Flow, dynamic_batching, requests
+
+
+class MyExecutor(Executor):
+    @requests(on=['/cat', '/kitten'])
+    @dynamic_batching(preferred_batch_size=4, timeout=2000)
+    def cat_fun(self, docs, **kwargs):
+        return DocumentArray([Document(text='cat')])
+
+    @requests()
+    @dynamic_batching(preferred_batch_size=10, timeout=10000)
+    def default_fun(self, docs, **kwargs):
+        return DocumentArray([Document(text='bar')])
+
+
+def test_dynamic_batching_config(tmpdir):
+    TMPDIR: Path = Path(tmpdir)
+    f = Flow(port=12345).add(
+        uses=MyExecutor,
+        name='exec0',
+        uses_dynamic_batching={'/foo': {'preferred_batch_size': 2, 'timeout': 4000}},
+    )
+    f.save_config(str(TMPDIR / 'flow0.yml'))
+
+    f1 = Flow.load_config(str(TMPDIR / 'flow0.yml'))
+    assert (
+        f._deployment_nodes['exec0'].args.uses_dynamic_batching
+        == f1._deployment_nodes['exec0'].args.uses_dynamic_batching
+    )
diff --git a/tests/unit/serve/executors/test_bad_executor_constructor.py b/tests/unit/serve/executors/test_bad_executor_constructor.py
index 15d993ea366cb..3815a910dcab4 100644
--- a/tests/unit/serve/executors/test_bad_executor_constructor.py
+++ b/tests/unit/serve/executors/test_bad_executor_constructor.py
@@ -13,7 +13,7 @@ def foo(self, **kwargs):
 
 
 class GoodExecutor2(Executor):
-    def __init__(self, metas, requests, runtime_args):
+    def __init__(self, metas, requests, runtime_args, dynamic_batching):
         pass
 
     @requests
@@ -24,7 +24,7 @@ def foo(self, docs, parameters, docs_matrix):
 def test_bad_executor_constructor():
     # executor can be used as out of Flow as Python object
     exec1 = GoodExecutor()
-    exec2 = GoodExecutor2({}, {}, {})
+    exec2 = GoodExecutor2({}, {}, {}, {})
 
     # can be used in the Flow
     with Flow().add(uses=GoodExecutor):
diff --git a/tests/unit/serve/executors/test_decorators.py b/tests/unit/serve/executors/test_decorators.py
index ebc6660564366..0dc67fe799886 100644
--- a/tests/unit/serve/executors/test_decorators.py
+++ b/tests/unit/serve/executors/test_decorators.py
@@ -4,7 +4,7 @@
 
 from jina.helper import iscoroutinefunction
 from jina.serve.executors import get_executor_taboo
-from jina.serve.executors.decorators import requests
+from jina.serve.executors.decorators import dynamic_batching, requests
 from jina.serve.helper import store_init_kwargs
 
 
@@ -63,3 +63,26 @@ async def fn_3(*args, **kwargs):
     assert not iscoroutinefunction(getattr(fn_2, 'fn'))
     assert hasattr(fn_3, 'fn')
     assert iscoroutinefunction(getattr(fn_3, 'fn'))
+
+
+def test_dynamic_batching():
+    @dynamic_batching()
+    def fn_2(*args, **kwargs):
+        pass
+
+    assert hasattr(fn_2, 'fn')
+
+
+def test_async_dynamic_batching():
+    @dynamic_batching
+    def fn_2(*args, **kwargs):
+        pass
+
+    @dynamic_batching
+    async def fn_3(*args, **kwargs):
+        pass
+
+    assert hasattr(fn_2, 'fn')
+    assert not iscoroutinefunction(getattr(fn_2, 'fn'))
+    assert hasattr(fn_3, 'fn')
+    assert iscoroutinefunction(getattr(fn_3, 'fn'))
diff --git a/tests/unit/serve/executors/test_executor.py b/tests/unit/serve/executors/test_executor.py
index a53b0ce918f6b..cdcbb1a6a89b9 100644
--- a/tests/unit/serve/executors/test_executor.py
+++ b/tests/unit/serve/executors/test_executor.py
@@ -9,15 +9,15 @@
 import yaml
 from docarray import Document, DocumentArray
 
-from jina import Client, Executor, Flow, __cache_path__, requests
+from jina import Client, Executor, Flow, __cache_path__, dynamic_batching, requests
 from jina.clients.request import request_generator
 from jina.excepts import RuntimeFailToStart
+from jina.helper import random_port
 from jina.parsers import set_pod_parser
 from jina.serve.executors.metas import get_default_metas
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
-from jina.helper import random_port
 
 
 class WorkspaceExec(Executor):
@@ -104,7 +104,7 @@ def test_executor_with_pymodule_path():
 
 def test_flow_uses_with_pymodule_path():
     with Flow.load_config(
-            '''
+        '''
     jtype: Flow
     executors:
         - uses: unit.serve.executors.dummy_executor.MyExecutor
@@ -115,13 +115,13 @@ def test_flow_uses_with_pymodule_path():
         pass
 
     with Flow().add(
-            uses='unit.serve.executors.dummy_executor.MyExecutor', uses_with={'bar': 123}
+        uses='unit.serve.executors.dummy_executor.MyExecutor', uses_with={'bar': 123}
     ):
         pass
 
     with pytest.raises(RuntimeFailToStart):
         with Flow.load_config(
-                '''
+            '''
             jtype: Flow
             executors:
                 - uses: jina.no_valide.executor
@@ -238,7 +238,7 @@ def test_executor_workspace(test_metas_workspace_replica_pods, shard_id):
 
 @pytest.mark.parametrize('shard_id', [None, -1], indirect=True)
 def test_executor_workspace_parent_replica_nopea(
-        test_metas_workspace_replica_pods, shard_id
+    test_metas_workspace_replica_pods, shard_id
 ):
     executor = Executor(
         metas={'name': test_metas_workspace_replica_pods['name']},
@@ -254,7 +254,7 @@ def test_executor_workspace_parent_replica_nopea(
 
 @pytest.mark.parametrize('shard_id', [0, 1, 2], indirect=True)
 def test_executor_workspace_parent_noreplica_pod(
-        test_metas_workspace_replica_pods, shard_id
+    test_metas_workspace_replica_pods, shard_id
 ):
     executor = Executor(
         metas={'name': test_metas_workspace_replica_pods['name']},
@@ -271,7 +271,7 @@ def test_executor_workspace_parent_noreplica_pod(
 
 @pytest.mark.parametrize('shard_id', [None, -1], indirect=True)
 def test_executor_workspace_parent_noreplica_nopea(
-        test_metas_workspace_replica_pods, shard_id
+    test_metas_workspace_replica_pods, shard_id
 ):
     executor = Executor(
         metas={'name': test_metas_workspace_replica_pods['name']},
@@ -406,7 +406,7 @@ def test_set_workspace(tmpdir):
         os.path.join(tmpdir, 'WorkspaceExec')
     )
     assert (
-            WorkspaceExec(workspace=str(tmpdir)).workspace == complete_workspace_no_replicas
+        WorkspaceExec(workspace=str(tmpdir)).workspace == complete_workspace_no_replicas
     )
 
 
@@ -451,10 +451,10 @@ def test_to_k8s_yaml(tmpdir, exec_type):
         with open(os.path.join(tmpdir, 'gateway', 'gateway.yml')) as f:
             gatewayyaml = list(yaml.safe_load_all(f))[-1]
             assert (
-                    gatewayyaml['spec']['template']['spec']['containers'][0]['ports'][0][
-                        'containerPort'
-                    ]
-                    == 2020
+                gatewayyaml['spec']['template']['spec']['containers'][0]['ports'][0][
+                    'containerPort'
+                ]
+                == 2020
             )
             gateway_args = gatewayyaml['spec']['template']['spec']['containers'][0][
                 'args'
@@ -560,7 +560,6 @@ def default_a(self, **kwargs):
             pass
 
     class B(A):
-
         @requests(on='/index')
         def b(self, **kwargs):
             pass
@@ -577,3 +576,67 @@ class C(B):
     assert set(C().requests.keys()) == {'/index', '/default', '_jina_dry_run_'}
     assert C().requests['/index'] == B.b
     assert C().requests['/default'] == A.default_a
+
+
+@pytest.mark.parametrize(
+    'inputs,expected_values',
+    [
+        (
+            dict(preferred_batch_size=4, timeout=5_000, max_batch_size=256),
+            dict(preferred_batch_size=4, timeout=5_000, max_batch_size=256),
+        ),
+        (
+            dict(preferred_batch_size=4, timeout=5_000),
+            dict(preferred_batch_size=4, timeout=5_000, max_batch_size=None),
+        ),
+        (
+            dict(preferred_batch_size=4, max_batch_size=256),
+            dict(preferred_batch_size=4, timeout=10_000, max_batch_size=256),
+        ),
+    ],
+)
+def test_dynamic_batching(inputs, expected_values):
+    class MyExec(Executor):
+        @dynamic_batching(**inputs)
+        def foo(self, docs, **kwargs):
+            pass
+
+    exec = MyExec()
+    assert exec.dynamic_batching['foo'] == expected_values
+
+
+@pytest.mark.parametrize(
+    'inputs,expected_values',
+    [
+        (
+            dict(preferred_batch_size=4, timeout=5_000, max_batch_size=256),
+            dict(preferred_batch_size=4, timeout=5_000, max_batch_size=256),
+        ),
+        (
+            dict(preferred_batch_size=4, timeout=5_000),
+            dict(preferred_batch_size=4, timeout=5_000, max_batch_size=None),
+        ),
+        (
+            dict(preferred_batch_size=4, max_batch_size=256),
+            dict(preferred_batch_size=4, timeout=10_000, max_batch_size=256),
+        ),
+    ],
+)
+def test_combined_decorators(inputs, expected_values):
+    class MyExecutor(Executor):
+        @dynamic_batching(**inputs)
+        @requests(on='/foo')
+        def foo(self, docs, **kwargs):
+            pass
+
+    exec = MyExecutor()
+    assert exec.dynamic_batching['foo'] == expected_values
+
+    class MyExecutor2(Executor):
+        @requests(on='/foo')
+        @dynamic_batching(**inputs)
+        def foo(self, docs, **kwargs):
+            pass
+
+    exec = MyExecutor2()
+    assert exec.dynamic_batching['foo'] == expected_values


Code review comments:
diff_hunk: @@ -0,0 +1,172 @@
+import asyncio
+from asyncio import Event, Task
+from typing import Any, Dict, List, Optional
+
+from jina import DocumentArray
+from jina.serve.executors import BaseExecutor
+from jina.types.request.data import DataRequest
+
+
+async def sleep_then_set(time_seconds: float, event: Event):
+    """Sleep for time_seconds and then set the event
+
+    :param time_seconds: time to sleep
+    :param event: event to set
+    """
+    await asyncio.sleep(time_seconds)
+    event.set()
+
+
+class BatchQueue:
+    """A batch queue that holds the data request and the executor."""
+
+    def __init__(
+        self,
+        executor: BaseExecutor,
+        exec_endpoint: str,
+        args: Any,
+        params: Optional[Dict] = None,
+        preferred_batch_size: int = 4,
+        timeout: int = 10_000,
+        max_batch_size: Optional[int] = None,
+    ) -> None:
+        if params is None:
+            params = dict()
+        self._executor: BaseExecutor = executor
+        self._exec_endpoint: str = exec_endpoint
+        self.args = args
+        self.params = params
+
+        self._preferred_batch_size: int = preferred_batch_size
+        self._timeout: int = timeout
+        self._max_batch_size: Optional[int] = max_batch_size
+        if self._max_batch_size:
+            if self._max_batch_size < self._preferred_batch_size:
+                raise ValueError(
+                    f'Max batch size must be greater than or equal to preferred batch size.'
+                    f'Got max batch size {self._max_batch_size} and preferred'
+                    f'batch size {self._preferred_batch_size}'
+                )
+
+        self._reset()
+
+    def __repr__(self) -> str:
+        return (
+            f'{self.__class__.__name__}({self._executor}, {self._exec_endpoint},'
+            f'preferred_batch_size={self._preferred_batch_size}, timeout={self._timeout})'
+        )
+
+    def __str__(self) -> str:
+        return self.__repr__()
+
+    def _reset(self) -> None:
+        """Set all events and reset the batch queue."""
+        self._requests: List[DataRequest] = []
+        self._request_lens: List[int] = []
+        self._big_doc: DocumentArray = DocumentArray.empty()
+
+        self._flush_trigger: Event = Event()
+        self._flush_task: Task = asyncio.create_task(
+            self.await_then_flush(self._flush_trigger)
+        )
path: jina/serve/runtimes/worker/batch_queue.py
text: @user1:
very tiny perf improvement but can also improve code:
I think the flush_task should only be created when there was a push to the batch queue. So it is reset to None and only started in `push`.
flush_task == None => no timer stared + no data in queue
flush_task != None => timer has already been started and there's data in queue.
We would be able to remove _timer_started

diff_hunk: @@ -0,0 +1,172 @@
+import asyncio
+from asyncio import Event, Task
+from typing import Any, Dict, List, Optional
+
+from jina import DocumentArray
+from jina.serve.executors import BaseExecutor
+from jina.types.request.data import DataRequest
+
+
+async def sleep_then_set(time_seconds: float, event: Event):
+    """Sleep for time_seconds and then set the event
+
+    :param time_seconds: time to sleep
+    :param event: event to set
+    """
+    await asyncio.sleep(time_seconds)
+    event.set()
+
+
+class BatchQueue:
+    """A batch queue that holds the data request and the executor."""
+
+    def __init__(
+        self,
+        executor: BaseExecutor,
+        exec_endpoint: str,
path: jina/serve/runtimes/worker/batch_queue.py
text: @user1:
Does the BatchQueue need to know about the executor ?
I think only the WorkerRequestHandler need to know about the BatchQueue.
The BatchQueue just needs to receive the function to execute (partial(executor.__acall__, exec_endpoint))

diff_hunk: @@ -0,0 +1,172 @@
+import asyncio
+from asyncio import Event, Task
+from typing import Any, Dict, List, Optional
+
+from jina import DocumentArray
+from jina.serve.executors import BaseExecutor
+from jina.types.request.data import DataRequest
+
+
+async def sleep_then_set(time_seconds: float, event: Event):
+    """Sleep for time_seconds and then set the event
+
+    :param time_seconds: time to sleep
+    :param event: event to set
+    """
+    await asyncio.sleep(time_seconds)
+    event.set()
+
+
+class BatchQueue:
+    """A batch queue that holds the data request and the executor."""
+
+    def __init__(
+        self,
+        executor: BaseExecutor,
+        exec_endpoint: str,
+        args: Any,
+        params: Optional[Dict] = None,
+        preferred_batch_size: int = 4,
+        timeout: int = 10_000,
+        max_batch_size: Optional[int] = None,
+    ) -> None:
+        if params is None:
+            params = dict()
+        self._executor: BaseExecutor = executor
+        self._exec_endpoint: str = exec_endpoint
+        self.args = args
+        self.params = params
+
+        self._preferred_batch_size: int = preferred_batch_size
+        self._timeout: int = timeout
+        self._max_batch_size: Optional[int] = max_batch_size
+        if self._max_batch_size:
+            if self._max_batch_size < self._preferred_batch_size:
+                raise ValueError(
+                    f'Max batch size must be greater than or equal to preferred batch size.'
+                    f'Got max batch size {self._max_batch_size} and preferred'
+                    f'batch size {self._preferred_batch_size}'
+                )
+
+        self._reset()
+
+    def __repr__(self) -> str:
+        return (
+            f'{self.__class__.__name__}({self._executor}, {self._exec_endpoint},'
+            f'preferred_batch_size={self._preferred_batch_size}, timeout={self._timeout})'
+        )
+
+    def __str__(self) -> str:
+        return self.__repr__()
+
+    def _reset(self) -> None:
+        """Set all events and reset the batch queue."""
+        self._requests: List[DataRequest] = []
+        self._request_lens: List[int] = []
+        self._big_doc: DocumentArray = DocumentArray.empty()
+
+        self._flush_trigger: Event = Event()
+        self._flush_task: Task = asyncio.create_task(
+            self.await_then_flush(self._flush_trigger)
+        )
+        self._timer_started: bool = False
+
+    def _start_timer(self):
+        asyncio.create_task(sleep_then_set(self._timeout / 1000, self._flush_trigger))
+        self._timer_started = True
+
+    async def push(self, request: DataRequest) -> Task:
+        """Append request to the queue. Once the request has been processed, the returned task will complete.
+
+        :param request: The request to append to the queue.
+
+        :return: The task that will be set once the request has been processed.
+        """
+        if self._max_batch_size:
+            if len(self._big_doc) + len(request.docs) > self._max_batch_size:
+                self._flush_trigger.set()
+                await self._flush_task
+
+        if not self._timer_started:
+            self._start_timer()
path: jina/serve/runtimes/worker/batch_queue.py
text: @user1:
according to what was suggested above, we can do:
```suggestion
        if not self. _flush_task:
            self._flush_task = asyncio.create_task(
                self.await_then_flush() # also taking in consideration another suggestion to remove parameter
            )
            asyncio.create_task(self.sleep_then_set()) # also taking in consideration another suggestion to remove parameter
```

diff_hunk: @@ -299,43 +355,63 @@ async def handle(
         :param tracing_context: Optional OpenTelemetry tracing context from the originating request.
         :returns: the processed message
         """
+
         # skip executor if endpoints mismatch
-        if (
-            requests[0].header.exec_endpoint not in self._executor.requests
-            and __default_endpoint__ not in self._executor.requests
-        ):
-            self.logger.debug(
-                f'skip executor: mismatch request, exec_endpoint: {requests[0].header.exec_endpoint}, requests: {self._executor.requests}'
-            )
-            return requests[0]
+        exec_endpoint: str = requests[0].header.exec_endpoint
+        if exec_endpoint not in self._executor.requests:
+            if __default_endpoint__ in self._executor.requests:
+                exec_endpoint = __default_endpoint__
+            else:
+                self.logger.debug(
+                    f'skip executor: mismatch request, exec_endpoint: {exec_endpoint}, requests: {self._executor.requests}'
+                )
+                return requests[0]
 
         self._record_request_size_monitoring(requests)
 
         params = self._parse_params(requests[0].parameters, self._executor.metas.name)
-        docs = WorkerRequestHandler.get_docs_from_request(
-            requests,
-            field='docs',
-        )
 
-        # executor logic
-        docs_matrix, docs_map = WorkerRequestHandler._get_docs_matrix_from_request(
-            requests
-        )
-        return_data = await self._executor.__acall__(
-            req_endpoint=requests[0].header.exec_endpoint,
-            docs=docs,
-            parameters=params,
-            docs_matrix=docs_matrix,
-            docs_map=docs_map,
-            tracing_context=tracing_context,
-        )
+        if exec_endpoint in self._batchqueue_config:
+            assert len(requests) == 1, 'dynamic batching does not support no_reduce'
+
+            param_key = str(params)
path: jina/serve/runtimes/worker/request_handling.py
text: @user1:
I don't think this is a reliable way to generate a key for parameters, try this for instance:
```python
d = {'a': 1, 'b': 'c'}
print(str(d)) # "{'a': 1, 'b': 'c'}"
d = {'b': 'c', 'a': 1}
print(str(d)) #  "{'b': 'c', 'a': 1}"
```
You can refer to this page for ways to hash a dict: https://stackoverflow.com/questions/5884066/hashing-a-dictionary
Ideally, `hash(frozenset(my_dict.items()))` would be great, if we can't use it, we'll have to use `json.dumps(d, sort_keys=True)`

@user2:
agree, this is not reliable

diff_hunk: @@ -299,43 +355,63 @@ async def handle(
         :param tracing_context: Optional OpenTelemetry tracing context from the originating request.
         :returns: the processed message
         """
+
         # skip executor if endpoints mismatch
-        if (
-            requests[0].header.exec_endpoint not in self._executor.requests
-            and __default_endpoint__ not in self._executor.requests
-        ):
-            self.logger.debug(
-                f'skip executor: mismatch request, exec_endpoint: {requests[0].header.exec_endpoint}, requests: {self._executor.requests}'
-            )
-            return requests[0]
+        exec_endpoint: str = requests[0].header.exec_endpoint
+        if exec_endpoint not in self._executor.requests:
+            if __default_endpoint__ in self._executor.requests:
+                exec_endpoint = __default_endpoint__
+            else:
+                self.logger.debug(
+                    f'skip executor: mismatch request, exec_endpoint: {exec_endpoint}, requests: {self._executor.requests}'
+                )
+                return requests[0]
 
         self._record_request_size_monitoring(requests)
 
         params = self._parse_params(requests[0].parameters, self._executor.metas.name)
-        docs = WorkerRequestHandler.get_docs_from_request(
-            requests,
-            field='docs',
-        )
 
-        # executor logic
-        docs_matrix, docs_map = WorkerRequestHandler._get_docs_matrix_from_request(
-            requests
-        )
-        return_data = await self._executor.__acall__(
-            req_endpoint=requests[0].header.exec_endpoint,
-            docs=docs,
-            parameters=params,
-            docs_matrix=docs_matrix,
-            docs_map=docs_map,
-            tracing_context=tracing_context,
-        )
+        if exec_endpoint in self._batchqueue_config:
+            assert len(requests) == 1, 'dynamic batching does not support no_reduce'
+
+            param_key = str(params)
+            if param_key not in self._batchqueue_instances[exec_endpoint]:
+                self._batchqueue_instances[exec_endpoint][param_key] = BatchQueue(
+                    executor=self._executor,
+                    exec_endpoint=exec_endpoint,
+                    args=self.args,
+                    params=params,
+                    **self._batchqueue_config[exec_endpoint],
path: jina/serve/runtimes/worker/request_handling.py
text: @user1:
as mentioned above, I believe here we should only pass a function to BatchQueue (partial(executor.__acall__, exec_endpoint))
This should help decouple both classes and make BatchQueue easier to test (no more need for a mock executor)

@user2:
Love it

diff_hunk: @@ -0,0 +1,172 @@
+import asyncio
+from asyncio import Event, Task
+from typing import Any, Dict, List, Optional
+
+from jina import DocumentArray
+from jina.serve.executors import BaseExecutor
+from jina.types.request.data import DataRequest
+
+
+async def sleep_then_set(time_seconds: float, event: Event):
+    """Sleep for time_seconds and then set the event
+
+    :param time_seconds: time to sleep
+    :param event: event to set
+    """
+    await asyncio.sleep(time_seconds)
+    event.set()
+
+
+class BatchQueue:
+    """A batch queue that holds the data request and the executor."""
+
+    def __init__(
+        self,
+        executor: BaseExecutor,
+        exec_endpoint: str,
+        args: Any,
+        params: Optional[Dict] = None,
+        preferred_batch_size: int = 4,
+        timeout: int = 10_000,
+        max_batch_size: Optional[int] = None,
+    ) -> None:
+        if params is None:
+            params = dict()
+        self._executor: BaseExecutor = executor
+        self._exec_endpoint: str = exec_endpoint
+        self.args = args
+        self.params = params
+
+        self._preferred_batch_size: int = preferred_batch_size
+        self._timeout: int = timeout
+        self._max_batch_size: Optional[int] = max_batch_size
+        if self._max_batch_size:
+            if self._max_batch_size < self._preferred_batch_size:
+                raise ValueError(
+                    f'Max batch size must be greater than or equal to preferred batch size.'
+                    f'Got max batch size {self._max_batch_size} and preferred'
+                    f'batch size {self._preferred_batch_size}'
+                )
+
+        self._reset()
+
+    def __repr__(self) -> str:
+        return (
+            f'{self.__class__.__name__}({self._executor}, {self._exec_endpoint},'
+            f'preferred_batch_size={self._preferred_batch_size}, timeout={self._timeout})'
+        )
+
+    def __str__(self) -> str:
+        return self.__repr__()
+
+    def _reset(self) -> None:
+        """Set all events and reset the batch queue."""
+        self._requests: List[DataRequest] = []
+        self._request_lens: List[int] = []
+        self._big_doc: DocumentArray = DocumentArray.empty()
+
+        self._flush_trigger: Event = Event()
+        self._flush_task: Task = asyncio.create_task(
+            self.await_then_flush(self._flush_trigger)
+        )
+        self._timer_started: bool = False
+
+    def _start_timer(self):
+        asyncio.create_task(sleep_then_set(self._timeout / 1000, self._flush_trigger))
+        self._timer_started = True
+
+    async def push(self, request: DataRequest) -> Task:
+        """Append request to the queue. Once the request has been processed, the returned task will complete.
+
+        :param request: The request to append to the queue.
+
+        :return: The task that will be set once the request has been processed.
+        """
+        if self._max_batch_size:
+            if len(self._big_doc) + len(request.docs) > self._max_batch_size:
+                self._flush_trigger.set()
+                await self._flush_task
+
+        if not self._timer_started:
+            self._start_timer()
+
+        docs = request.docs
+        self._big_doc.extend(docs)
+        self._requests.append(request)
+        self._request_lens.append(len(docs))
+
+        if len(self._big_doc) >= self._preferred_batch_size:
+            self._flush_trigger.set()
+
+        return self._flush_task
+
+    async def await_then_flush(self, trigger_event: Event) -> None:
path: jina/serve/runtimes/worker/batch_queue.py
text: @user1:
since the trigger_event is always self._flush_trigger, I think just always using it and removing the parameter improves readability

diff_hunk: @@ -0,0 +1,172 @@
+import asyncio
+from asyncio import Event, Task
+from typing import Any, Dict, List, Optional
+
+from jina import DocumentArray
+from jina.serve.executors import BaseExecutor
+from jina.types.request.data import DataRequest
+
+
+async def sleep_then_set(time_seconds: float, event: Event):
path: jina/serve/runtimes/worker/batch_queue.py
text: @user1:
let's make it private

diff_hunk: @@ -0,0 +1,172 @@
+import asyncio
+from asyncio import Event, Task
+from typing import Any, Dict, List, Optional
+
+from jina import DocumentArray
+from jina.serve.executors import BaseExecutor
+from jina.types.request.data import DataRequest
+
+
+async def sleep_then_set(time_seconds: float, event: Event):
+    """Sleep for time_seconds and then set the event
+
+    :param time_seconds: time to sleep
+    :param event: event to set
+    """
+    await asyncio.sleep(time_seconds)
+    event.set()
+
+
+class BatchQueue:
+    """A batch queue that holds the data request and the executor."""
+
+    def __init__(
+        self,
+        executor: BaseExecutor,
+        exec_endpoint: str,
+        args: Any,
+        params: Optional[Dict] = None,
+        preferred_batch_size: int = 4,
+        timeout: int = 10_000,
+        max_batch_size: Optional[int] = None,
+    ) -> None:
+        if params is None:
+            params = dict()
+        self._executor: BaseExecutor = executor
+        self._exec_endpoint: str = exec_endpoint
+        self.args = args
+        self.params = params
+
+        self._preferred_batch_size: int = preferred_batch_size
+        self._timeout: int = timeout
+        self._max_batch_size: Optional[int] = max_batch_size
+        if self._max_batch_size:
+            if self._max_batch_size < self._preferred_batch_size:
+                raise ValueError(
+                    f'Max batch size must be greater than or equal to preferred batch size.'
+                    f'Got max batch size {self._max_batch_size} and preferred'
+                    f'batch size {self._preferred_batch_size}'
+                )
+
+        self._reset()
+
+    def __repr__(self) -> str:
+        return (
+            f'{self.__class__.__name__}({self._executor}, {self._exec_endpoint},'
+            f'preferred_batch_size={self._preferred_batch_size}, timeout={self._timeout})'
+        )
+
+    def __str__(self) -> str:
+        return self.__repr__()
+
+    def _reset(self) -> None:
+        """Set all events and reset the batch queue."""
+        self._requests: List[DataRequest] = []
+        self._request_lens: List[int] = []
+        self._big_doc: DocumentArray = DocumentArray.empty()
+
+        self._flush_trigger: Event = Event()
+        self._flush_task: Task = asyncio.create_task(
+            self.await_then_flush(self._flush_trigger)
+        )
+        self._timer_started: bool = False
+
+    def _start_timer(self):
+        asyncio.create_task(sleep_then_set(self._timeout / 1000, self._flush_trigger))
path: jina/serve/runtimes/worker/batch_queue.py
text: @user1:
I suggest simplifying code here by making sleep_then_set a private method of BatchQueue that directly uses timeout and flush_trigger (no need to receive parameters)

@user1:
maybe do this just for self._timeout, because self._flush_trigger is renewed on every flush and we don't want old timers to set triggers for new flushes