Repository: jina-ai/serve
Pull request: feat: disable prefetch by default, remove on-recv :boom: 
# Streamer
Implements request streaming in 2 ways

- prefetch - controls speed of request sending, handles overloading https://github.com/jina-ai/jina/blob/d9b841474e29ffa2d332f006fc2d773881a1b6ae/jina/peapods/runtimes/stream/mixin.py#L108-L110
- no-prefetch - async send/receive requests without control https://github.com/jina-ai/jina/blob/d9b841474e29ffa2d332f006fc2d773881a1b6ae/jina/peapods/runtimes/stream/mixin.py#L52-L54

  ### GatewayStreamer
  It handles `zmq`/`gRPC` requests for `gateway -> Executors -> gateway`  communications. `prefetch` is used to control load on Executors from gateway.  Prefetching can be disabled by setting `prefetch=0` in the `Flow`. 
  
  ### ClientStreamer
  It handles `HTTP`/`Websocket` requests for  `client  -> gateway -> client` communications. `prefetch` is not used here as there's no load to control.

# Requests handling at gateway

### `http` protocol

- Each request from the Client (received as json) is converted to a `request_generator` of len(1) at the gateway.
- `GatewayStreamer` is invoked with this generator. Since `len(generator)` is always 1, there's no prefetching to do even when `prefetch` is set with the Flow. It always sends/receives 1 request to/from Executor in a stream.
- Since `HTTPGateway` doesn't implement streaming a gateway, each request creates a new stream. One/multiple clients can perform parallel requests (we don't control the load for this).


### `websocket` protocol
- Client iterates through the `request_generator` and sends each request as bytes to the gateway. Gateway receives sequence of requests as bytes and converts them into a new generator (len >= 1).  `GatewayStreamer` is invoked with this. It can optionally use `prefetch`. 
- It is important to note that, the request_generator is traversed twice - 
    - at Client (`Document -> request_generator -> bytestream`) (`Document` creation happens here)
    - at Gateway (`bytestream -> request_generator`).


### `grpc` protocol
gRPC Client internally handles sending `request_generator` from Client to gateway. (`Streamer` implementation is not used here). `request_generator` is traversed at gateway (either with prefetcher or no-prefetch) and response is streamed back to the Client.
 

## `prefetch` vs `no-prefetch` tests

1. Slow Client & Fast Executor. Validates `input -> executor -> on_done` for each request in sequence.

  https://github.com/jina-ai/jina/blob/d402347b6bf30d63aa356c62de57aa05b075cd80/tests/integration/streaming/test_clients_streaming.py#L94-L96

2. Fast Client & Slow Executor. Validates requests are non-blocking. 

  https://github.com/jina-ai/jina/blob/d402347b6bf30d63aa356c62de57aa05b075cd80/tests/integration/streaming/test_clients_streaming.py#L143-L145

3. Concurrent clients with & without prefetch.

https://github.com/jina-ai/jina/blob/008795a21d8eccdcb9d89b5f9299411cae83236c/tests/integration/streaming/test_clients_streaming.py#L223


Problem statement:
Improve prefetching
If an async client is slow to yield requests (with prefetching > 1) this will also slow done the response streaming. Requests are processed by the flow as they come in via the async iterator, but they results are only streamed back to the client once the input iterator is complete or the prefetching limit has been hit.

Hints from issue discussion:

Commit message: refactor: set prefetch to 1 by default

Patch to review:
diff --git a/jina/parsers/peapods/runtimes/remote.py b/jina/parsers/peapods/runtimes/remote.py
index 838798283ca93..b10b53a67e715 100644
--- a/jina/parsers/peapods/runtimes/remote.py
+++ b/jina/parsers/peapods/runtimes/remote.py
@@ -164,14 +164,8 @@ def mixin_prefetch_parser(parser=None):
     gp.add_argument(
         '--prefetch',
         type=int,
-        default=50,
-        help='The number of pre-fetched requests from the client',
-    )
-    gp.add_argument(
-        '--prefetch-on-recv',
-        type=int,
         default=1,
-        help='The number of additional requests to fetch on every receive',
+        help='The number of pre-fetched requests from the client',
     )
 
 
diff --git a/jina/peapods/runtimes/prefetch/base.py b/jina/peapods/runtimes/prefetch/base.py
index 4357210cb003b..407a39658dd98 100644
--- a/jina/peapods/runtimes/prefetch/base.py
+++ b/jina/peapods/runtimes/prefetch/base.py
@@ -138,9 +138,7 @@ async def prefetch_req(
                 for r in asyncio.as_completed(prefetch_task):
                     yield await r
                     if not is_req_empty:
-                        is_req_empty = await prefetch_req(
-                            self.args.prefetch_on_recv, onrecv_task
-                        )
+                        is_req_empty = await prefetch_req(1, onrecv_task)
 
                 # this list dries, clear it and feed it with on_recv_task
                 prefetch_task.clear()


Code review comments:
diff_hunk: @@ -164,14 +164,8 @@ def mixin_prefetch_parser(parser=None):
     gp.add_argument(
         '--prefetch',
         type=int,
-        default=50,
-        help='The number of pre-fetched requests from the client',
-    )
-    gp.add_argument(
-        '--prefetch-on-recv',
-        type=int,
         default=1,
-        help='The number of additional requests to fetch on every receive',
+        help='The number of pre-fetched requests from the client',
path: jina/parsers/peapods/runtimes/remote.py
text: @user1:
```suggestion
        help='The number of requests fetched from the client before feeding to the first Executor',
```