Repository: jina-ai/serve (language: python)

Issue:
Thread lock the Jina client (or streamer) using RLock
The Jina GRPCClient cannot be used in multiple threads (due to a limitation from grpc: https://github.com/grpc/grpc/issues/25364).

Use [RLock](https://docs.python.org/3/library/threading.html#rlock-objects) to restrict the usage of grpc client within 1 thread at a time and test this in a multi threaded setup (flask).
You can get inspired by this commit https://github.com/jina-ai/jina/commit/af988e9920b6ff35fb3014f32b9b27d86797bd3b


Discussion / hints:
Why an `RLock` and not a simple `Lock`?
> Why an `RLock` and not a simple `Lock`?

An RLock let's you re acquire the lock from the same thread, therefore it shouldn't harm our async support.
A simple lock means only 1 request can be made at a time
I see

Fix patch:
diff --git a/docs/concepts/client/index.md b/docs/concepts/client/index.md
index 589ac2cd88ff9..859203d046c61 100644
--- a/docs/concepts/client/index.md
+++ b/docs/concepts/client/index.md
@@ -150,15 +150,14 @@ Client(host='https://my.awesome.flow:1234', port=4321)
 
 ````{admonition} Caution
 :class: caution
-In case you instanciate a `Client` object using the `grpc` protocol, keep in mind that `grpc` clients cannot be used in 
-a multi-threaded environment (check [this gRPC issue](https://github.com/grpc/grpc/issues/25364) for reference).
+
+We apply `RLock` to avoid [this gRPC issue](https://github.com/grpc/grpc/issues/25364), so that `grpc` clients can be used in a multi-threaded environment.
+
 What you should do, is to rely on asynchronous programming or multi-processing rather than multi-threading.
 For instance, if you're building a web server, you can introduce multi-processing based parallelism to your app using 
 `gunicorn`: `gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker ...`
 ````
 
-
-
 (client-compress)=
 ## Enable compression
 
diff --git a/jina/clients/base/grpc.py b/jina/clients/base/grpc.py
index bf45dc63126a3..0d96c8b947bc3 100644
--- a/jina/clients/base/grpc.py
+++ b/jina/clients/base/grpc.py
@@ -1,10 +1,10 @@
 import asyncio
 import json
+import threading
 from typing import TYPE_CHECKING, Optional, Tuple
 
 import grpc
 from grpc import RpcError
-
 from jina.clients.base import BaseClient
 from jina.clients.helper import callback_exec
 from jina.excepts import BadClientInput, BadServerFlow, InternalNetworkError
@@ -24,6 +24,8 @@ class GRPCBaseClient(BaseClient):
 
     It manages the asyncio event loop internally, so all interfaces are synchronous from the outside.
     """
+    
+    _lock = threading.RLock()
 
     async def _is_flow_ready(self, **kwargs) -> bool:
         """Sends a dry run to the Flow to validate if the Flow is ready to receive requests
@@ -203,81 +205,82 @@ async def _get_results(
                 metadata = metadata or ()
                 metadata = metadata + (('__results_in_order__', 'true'),)
 
-            async with GrpcConnectionPool.get_grpc_channel(
-                    f'{self.args.host}:{self.args.port}',
-                    options=options,
-                    asyncio=True,
-                    tls=self.args.tls,
-                    aio_tracing_client_interceptors=self.aio_tracing_client_interceptors(),
-            ) as channel:
-                self.logger.debug(f'connected to {self.args.host}:{self.args.port}')
+            with self._lock:
+                async with GrpcConnectionPool.get_grpc_channel(
+                        f'{self.args.host}:{self.args.port}',
+                        options=options,
+                        asyncio=True,
+                        tls=self.args.tls,
+                        aio_tracing_client_interceptors=self.aio_tracing_client_interceptors(),
+                ) as channel:
+                    self.logger.debug(f'connected to {self.args.host}:{self.args.port}')
 
-                with ProgressBar(
-                        total_length=self._inputs_length, disable=not self.show_progress
-                ) as p_bar:
-                    try:
-                        if stream:
-                            async for resp in self._stream_rpc(
-                                    channel=channel,
-                                    req_iter=req_iter,
-                                    metadata=metadata,
-                                    on_error=on_error,
-                                    on_done=on_done,
-                                    on_always=on_always,
-                                    continue_on_error=continue_on_error,
-                                    p_bar=p_bar,
-                                    **kwargs,
-                            ):
-                                yield resp
-                        else:
-                            async for resp in self._unary_rpc(
-                                    channel=channel,
-                                    req_iter=req_iter,
-                                    metadata=metadata,
-                                    on_error=on_error,
-                                    on_done=on_done,
-                                    on_always=on_always,
-                                    continue_on_error=continue_on_error,
-                                    p_bar=p_bar,
-                                    results_in_order=results_in_order,
-                                    **kwargs,
-                            ):
-                                yield resp
+                    with ProgressBar(
+                            total_length=self._inputs_length, disable=not self.show_progress
+                    ) as p_bar:
+                        try:
+                            if stream:
+                                async for resp in self._stream_rpc(
+                                        channel=channel,
+                                        req_iter=req_iter,
+                                        metadata=metadata,
+                                        on_error=on_error,
+                                        on_done=on_done,
+                                        on_always=on_always,
+                                        continue_on_error=continue_on_error,
+                                        p_bar=p_bar,
+                                        **kwargs,
+                                ):
+                                    yield resp
+                            else:
+                                async for resp in self._unary_rpc(
+                                        channel=channel,
+                                        req_iter=req_iter,
+                                        metadata=metadata,
+                                        on_error=on_error,
+                                        on_done=on_done,
+                                        on_always=on_always,
+                                        continue_on_error=continue_on_error,
+                                        p_bar=p_bar,
+                                        results_in_order=results_in_order,
+                                        **kwargs,
+                                ):
+                                    yield resp
 
-                    except (grpc.aio._call.AioRpcError, InternalNetworkError) as err:
-                        my_code = err.code()
-                        my_details = err.details()
-                        trailing_metadata = extract_trailing_metadata(err)
-                        msg = f'gRPC error: {my_code} {my_details}'
-                        if trailing_metadata:
-                            msg = f'gRPC error: {my_code} {my_details}\n{trailing_metadata}'
+                        except (grpc.aio._call.AioRpcError, InternalNetworkError) as err:
+                            my_code = err.code()
+                            my_details = err.details()
+                            trailing_metadata = extract_trailing_metadata(err)
+                            msg = f'gRPC error: {my_code} {my_details}'
+                            if trailing_metadata:
+                                msg = f'gRPC error: {my_code} {my_details}\n{trailing_metadata}'
 
-                        if my_code == grpc.StatusCode.UNAVAILABLE:
-                            self.logger.error(
-                                f'{msg}\nThe ongoing request is terminated as the server is not available or closed already.'
-                            )
-                            raise ConnectionError(my_details)
-                        elif my_code == grpc.StatusCode.DEADLINE_EXCEEDED:
-                            self.logger.error(
-                                f'{msg}\nThe ongoing request is terminated due to a server-side timeout.'
-                            )
-                            raise ConnectionError(my_details)
-                        elif my_code == grpc.StatusCode.INTERNAL:
-                            self.logger.error(
-                                f'{msg}\ninternal error on the server side'
-                            )
-                            raise err
-                        elif (
-                                my_code == grpc.StatusCode.UNKNOWN
-                                and 'asyncio.exceptions.TimeoutError' in my_details
-                        ):
-                            raise BadClientInput(
-                                f'{msg}\n'
-                                'often the case is that you define/send a bad input iterator to jina, '
-                                'please double check your input iterator'
-                            ) from err
-                        else:
-                            raise BadServerFlow(msg) from err
+                            if my_code == grpc.StatusCode.UNAVAILABLE:
+                                self.logger.error(
+                                    f'{msg}\nThe ongoing request is terminated as the server is not available or closed already.'
+                                )
+                                raise ConnectionError(my_details)
+                            elif my_code == grpc.StatusCode.DEADLINE_EXCEEDED:
+                                self.logger.error(
+                                    f'{msg}\nThe ongoing request is terminated due to a server-side timeout.'
+                                )
+                                raise ConnectionError(my_details)
+                            elif my_code == grpc.StatusCode.INTERNAL:
+                                self.logger.error(
+                                    f'{msg}\ninternal error on the server side'
+                                )
+                                raise err
+                            elif (
+                                    my_code == grpc.StatusCode.UNKNOWN
+                                    and 'asyncio.exceptions.TimeoutError' in my_details
+                            ):
+                                raise BadClientInput(
+                                    f'{msg}\n'
+                                    'often the case is that you define/send a bad input iterator to jina, '
+                                    'please double check your input iterator'
+                                ) from err
+                            else:
+                                raise BadServerFlow(msg) from err
 
         except KeyboardInterrupt:
             self.logger.warning('user cancel the process')


Test patch:
diff --git a/tests/integration/concurrent_clients/test_multithread_client.py b/tests/integration/concurrent_clients/test_multithread_client.py
new file mode 100644
index 0000000000000..97ee5a59a5873
--- /dev/null
+++ b/tests/integration/concurrent_clients/test_multithread_client.py
@@ -0,0 +1,57 @@
+
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+from jina import Client, Document, Executor, Flow, requests
+from jina.helper import random_port
+
+
+class MyExecutor(Executor):
+    @requests
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'I am coming from MyExecutor'
+                
+def flow_run(flow, stop_event):
+    with flow:
+        flow.block(stop_event)
+
+
+def client_post(doc, client):
+    result = client.post(on='/', inputs=doc)[0]
+    return result
+
+
+def test_multithread_client(capsys):
+    port = random_port()
+    stop_event = threading.Event()
+    flow = Flow(port=port).add(uses=MyExecutor)
+    t = threading.Thread(target=flow_run, args=(flow, stop_event))
+
+    c = Client(port=port)
+
+    try:
+        with capsys.disabled():
+            t.start()
+            time.sleep(5)
+
+        with ThreadPoolExecutor(max_workers=50) as pool:
+            tasks = []
+            for i in range(1000):
+                doc = Document(id=f'{i}', text='hello world')
+                task = pool.submit(client_post, doc, c)
+                tasks.append(task)
+
+            for i,task in enumerate(tasks):
+                result = task.result()
+                assert result.id == f'{i}'
+                assert result.text == 'I am coming from MyExecutor'
+
+        with capsys.disabled():
+            stdout, stderr = capsys.readouterr()
+            assert 'BlockingIOError' not in stderr
+
+    finally:
+        stop_event.set()
+        t.join()