Repository: jina-ai/serve
Pull request: fix: close loop from run_async
We are leaking `asyncio` loops in `run_ascynio`. We use `get_or_reuse_loop()` there to create a new `asyncio` loop in each call. This loop is then used in `run_until_complete`. This completes the coroutine, but does NOT close the loop.

To fix the issue this PR now gets every existing loop. Not only the running loop as before. This will correctly reuse previously created loops. If there is no loop or of it is closed already, we create a new loop.

I also add `psutil` as a new dependency for testing. I could also do its purpose manually, but it seems more convenient to use this library here and I dont think we need to be too restrictive for CI purposes?

Fixes: https://github.com/jina-ai/jina/issues/4725

Problem statement:
Client.encode() does not close file descriptors used.
I am calling .encode() multiple times and it seems as though file descriptors are being opened but not closed which results in this error after running for a period of time.

```
E0429 15:10:27.741673000 4681657856 wakeup_fd_pipe.cc:39]              pipe creation failed (24): Too many open files
E0429 15:10:27.744256000 4681657856 ev_poll_posix.cc:1415]             Skipping poll because of no wakeup fd.
E0429 15:10:27.744857000 4681657856 ev_posix.cc:228]                   No event engine could be initialized from all
```

when running `lsof | wc -l` I start to get crazy numbers of open files after a while.

You can run this test code with clip-server running to expose leak as fast as possible:

```python
from clip_client import Client


c = Client("grpc://localhost:51000")

while True:
   c.encode([" "]) 
```

Hints from issue discussion:
I can confirm this is a issue of Jina, the backend of CLIP-as-service, the minimum reproducible example is the following:
```python
import os
import subprocess

from jina import Flow

pid = os.getpid()
print('pid', pid)
print('before', subprocess.getoutput(f'lsof -p {pid} | wc -l'))
with Flow() as f:
    for _ in range(1000):
        f.post('/')
    print('after', subprocess.getoutput(f'lsof -p {pid} | wc -l'))

```

which gives:

```
pid 2679
before      154
⠙ 1/1 waiting  to be ready...
╭───────── 🎉 Flow is ready to serve! ──────────╮
│  🔗         Protocol                  GRPC    │
│  🏠     Local access         0.0.0.0:49458    │
│  🔒  Private network  192.168.31.160:49458    │
│  🌍   Public address  217.70.138.123:49458    │
╰───────────────────────────────────────────────╯
after     5194

```

from 154 -> 5194. I tried different repeatition of `.post` seems like the relationship is:

**Each request open 5 file handler, and there are constant 40 handlers always being opened.**

| For loop | Before | After | Delta |
| --- | --- | --- |  --- |
| 1 |  154 | 199 |  45 | 
| 10 |  154 | 244 | 90 |
| 100 | 154 | 694 |  540 |
| 1000 | 154 | 5194 | 5040 |


Also the following variants do not make any difference, the number is the same as table above:
```python
import os
import subprocess

from jina import Flow, Client

pid = os.getpid()
print('pid', pid)
print('before', subprocess.getoutput(f'lsof -p {pid} | wc -l'))
with Flow() as f:
    c = Client(port=f.port)
    for _ in range(1000):
        c.post('/')
    print('after', subprocess.getoutput(f'lsof -p {pid} | wc -l'))

```

```python
import os
import subprocess

from jina import Flow, Client

pid = os.getpid()
print('pid', pid)
print('before', subprocess.getoutput(f'lsof -p {pid} | wc -l'))
with Flow() as f:
    
    for _ in range(1000):
        c = Client(port=f.port)
        c.post('/')
    print('after', subprocess.getoutput(f'lsof -p {pid} | wc -l'))

```

Moving to upstream for better ownership. Thanks for the report!
A better visualization on diff

```python
import os
import subprocess
import sys
import webbrowser
from difflib import HtmlDiff

from jina import Flow, Client

pid = os.getpid()
print('pid', pid)
before = subprocess.getoutput(f'lsof -p {pid}')
with Flow() as f:
    c = Client(port=f.port)
    for _ in range(10):
        c.post('/')
    after = subprocess.getoutput(f'lsof -p {pid}')

with open('diff.html', 'w') as fp:
    fp.writelines(HtmlDiff().make_file(before.split('\n'), after.split('\n')))

webbrowser.open('file://' + os.path.realpath('diff.html'), new=2)


```




<img width="1616" alt="image" src="https://user-images.githubusercontent.com/2041322/166095488-a07ba663-9454-4145-9b36-fb377872a950.png">

<img width="909" alt="image" src="https://user-images.githubusercontent.com/2041322/166095996-16fa3534-2d51-4cf6-8a6c-d9ab30eb32be.png">

May or may not be related

https://github.com/QuarkChain/pyquarkchain/issues/411#issuecomment-484720553
https://github.com/python/cpython/issues/77262#issuecomment-1093779482
@hanxiao Thank you so much for your efforts! Hope this gets fixed soon.
I think our `run_async` helper function is at fault here. I got an even simpler mwu to reproduce with this:

```
import psutil

from jina.helper import run_async


async def do_something():
    pass

for i in range(1):
    run_async(do_something)

p = psutil.Process()
print(p.num_fds())
```
A workaround for now is to use the Client in its [async version](https://docs.jina.ai/fundamentals/flow/client/#async-python-client)

In this case we dont use `run_async` and there should be no file descriptors being leaked
> A workaround for now is to use the Client in its [async version](https://docs.jina.ai/fundamentals/flow/client/#async-python-client)
> 
> In this case we dont use `run_async` and there should be no file descriptors being leaked

Thank you, I will try that. Thank you for your help!

Commit message: fix: create loop if not existing

Patch to review:
diff --git a/extra-requirements.txt b/extra-requirements.txt
index c78151378af4e..c574f4d92a656 100644
--- a/extra-requirements.txt
+++ b/extra-requirements.txt
@@ -76,3 +76,4 @@ pytest-kind==21.1.3:        test
 pytest-lazy-fixture:        test
 sgqlc:                      cicd, graphql
 prometheus_client:          perf
+psutil:                     test
diff --git a/jina/helper.py b/jina/helper.py
index 5baffbcbf7174..a2b53d1e22ade 100644
--- a/jina/helper.py
+++ b/jina/helper.py
@@ -993,7 +993,8 @@ def _update_policy():
         try:
             import uvloop
 
-            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+            if not isinstance(asyncio.get_event_loop_policy(), uvloop.EventLoopPolicy):
+                asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
         except ModuleNotFoundError:
             warnings.warn(
                 'Install `uvloop` via `pip install "jina[uvloop]"` for better performance.'
@@ -1006,13 +1007,13 @@ def get_or_reuse_loop():
 
     :return: A new eventloop or reuse the current opened eventloop.
     """
+    _update_policy()
     try:
-        loop = asyncio.get_running_loop()
+        loop = asyncio.get_event_loop()
         if loop.is_closed():
             raise RuntimeError
     except RuntimeError:
-        _update_policy()
-        # no running event loop
+        # no event loop
         # create a new loop
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
diff --git a/jina/resources/extra-requirements.txt b/jina/resources/extra-requirements.txt
index c78151378af4e..c574f4d92a656 100644
--- a/jina/resources/extra-requirements.txt
+++ b/jina/resources/extra-requirements.txt
@@ -76,3 +76,4 @@ pytest-kind==21.1.3:        test
 pytest-lazy-fixture:        test
 sgqlc:                      cicd, graphql
 prometheus_client:          perf
+psutil:                     test
diff --git a/tests/unit/test_helper.py b/tests/unit/test_helper.py
index 412a4b35c4687..73b5416fef7e3 100644
--- a/tests/unit/test_helper.py
+++ b/tests/unit/test_helper.py
@@ -2,6 +2,7 @@
 from types import SimpleNamespace
 
 import numpy as np
+import psutil
 import pytest
 
 from jina import Executor, Flow, __default_endpoint__
@@ -19,6 +20,7 @@
     random_port,
     reset_ports,
     retry,
+    run_async,
 )
 from jina.hubble.helper import _get_hubble_base_url
 from jina.jaml.helper import complete_path
@@ -352,3 +354,19 @@ def test_port_free(port_generator):
     port = port_generator()
     is_free = is_port_free('localhost', port)
     assert is_free
+
+
+def test_run_async():
+    async def dummy():
+        pass
+
+    p = psutil.Process()
+
+    run_async(dummy)
+    first_fd_count = p.num_fds()
+
+    for i in range(10):
+        run_async(dummy)
+
+    end_fd_count = p.num_fds()
+    assert first_fd_count == end_fd_count


Code review comments:
diff_hunk: @@ -1006,13 +1007,13 @@ def get_or_reuse_loop():
 
     :return: A new eventloop or reuse the current opened eventloop.
     """
+    _update_policy()
     try:
-        loop = asyncio.get_running_loop()
+        loop = asyncio.get_event_loop()
path: jina/helper.py
text: @author:
this is the relevant change. We should use any loop (unless closed as checked in the next line).