We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Ray async api is not working with uvloop.
<!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->

### What is the problem?

Current Ray async api uses asyncio event loop's internal attribute to identify if the loop is running in the current current thread.

```python3
    loop = asyncio.get_event_loop()
    if loop.is_running():
        if loop._thread_id != threading.get_ident():
            # If the loop is runing outside current thread, we actually need
            # to do this to make sure the context is initialized.
            asyncio.run_coroutine_threadsafe(_async_init(), loop=loop)
```
This causes a problem when we uses Ray APIs inside Fast API because Fast API uses uvloop as its main event loop, and uvloop doesn't have `_thread_id` attribute.

@simon-mo Any good idea to fix this? It doesn't seem to be trivial. What about we do async_init() whenever asyncio loop is created in a different thread instead of checking if the event loop's thread id? I assume the only use case where asyncio loop is defined in a different thread is only inside async actor? 

### Reproduction (REQUIRED)
Please provide a script that can be run to reproduce the issue. The script should have **no external library dependencies** (i.e., use fake or mock data / environments):

```python3
import time
import asyncio 
​
import ray
import psutil
from fastapi import FastAPI, APIRouter
​
​
app = FastAPI(
    title="API template",
    description="Template to build upon for API serving and distributed computation",
    version="0.1.0",
    openapi_url="/openapi.json",
    docs_url="/docs",
)
​
@app.on_event("startup")
def startup_event():
    ray.init(num_cpus=2)
​
​
​
@app.on_event("shutdown")
def shutdown_event():
    ray.shutdown()
​
​
@app.get('/async')
async def non_seq_async_process():
    """
    async distributed execution
    """
    @ray.remote
    def slow_function(i):
        time.sleep(i)
        return i
​
    start_time = time.time()
​
    # result_ids = []
    # for i in range(10, 60, 10):
    #     result_ids.append(slow_function.remote(i))
    
    # results = ray.get(result_ids)
​
    results = await asyncio.wait([slow_function.remote(i) for i in range(10, 60, 10)])
​
    
    duration = time.time() - start_time
    out = "Executing the for loop took {:.3f} seconds.\n".format(duration)
    out += f"The results are: {results}\n"

```

If we cannot run your script, we cannot fix your issue.

- [x] I have verified my script runs in a clean environment and reproduces the issue.
- [x] I have verified the issue also occurs with the [latest wheels](https://docs.ray.io/en/latest/installation.html).


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/experimental/async_api.py`
Content:
```
1 # Note: asyncio is only compatible with Python 3
2 
3 import asyncio
4 import threading
5 
6 import ray
7 from ray.experimental.async_plasma import PlasmaEventHandler
8 from ray.services import logger
9 
10 handler = None
11 
12 
13 async def _async_init():
14     global handler
15     if handler is None:
16         worker = ray.worker.global_worker
17         loop = asyncio.get_event_loop()
18         handler = PlasmaEventHandler(loop, worker)
19         worker.core_worker.set_plasma_added_callback(handler)
20         logger.debug("AsyncPlasma Connection Created!")
21 
22 
23 def init():
24     """
25     Initialize synchronously.
26     """
27     assert ray.is_initialized(), "Please call ray.init before async_api.init"
28 
29     # Noop when handler is set.
30     if handler is not None:
31         return
32 
33     loop = asyncio.get_event_loop()
34     if loop.is_running():
35         if loop._thread_id != threading.get_ident():
36             # If the loop is runing outside current thread, we actually need
37             # to do this to make sure the context is initialized.
38             asyncio.run_coroutine_threadsafe(_async_init(), loop=loop)
39         else:
40             async_init_done = asyncio.get_event_loop().create_task(
41                 _async_init())
42             # Block until the async init finishes.
43             async_init_done.done()
44     else:
45         asyncio.get_event_loop().run_until_complete(_async_init())
46 
47 
48 def as_future(object_id):
49     """Turn an object_id into a Future object.
50 
51     Args:
52         object_id: A Ray object_id.
53 
54     Returns:
55         PlasmaObjectFuture: A future object that waits the object_id.
56     """
57     if handler is None:
58         init()
59     return handler.as_future(object_id)
60 
61 
62 def shutdown():
63     """Manually shutdown the async API.
64 
65     Cancels all related tasks and all the socket transportation.
66     """
67     global handler
68     if handler is not None:
69         handler.close()
70         handler = None
71
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/ray/experimental/async_api.py b/python/ray/experimental/async_api.py
--- a/python/ray/experimental/async_api.py
+++ b/python/ray/experimental/async_api.py
@@ -1,7 +1,4 @@
-# Note: asyncio is only compatible with Python 3
-
 import asyncio
-import threading
 
 import ray
 from ray.experimental.async_plasma import PlasmaEventHandler
@@ -10,7 +7,10 @@
 handler = None
 
 
-async def _async_init():
+def init():
+    """Initialize plasma event handlers for asyncio support."""
+    assert ray.is_initialized(), "Please call ray.init before async_api.init"
+
     global handler
     if handler is None:
         worker = ray.worker.global_worker
@@ -20,31 +20,6 @@
         logger.debug("AsyncPlasma Connection Created!")
 
 
-def init():
-    """
-    Initialize synchronously.
-    """
-    assert ray.is_initialized(), "Please call ray.init before async_api.init"
-
-    # Noop when handler is set.
-    if handler is not None:
-        return
-
-    loop = asyncio.get_event_loop()
-    if loop.is_running():
-        if loop._thread_id != threading.get_ident():
-            # If the loop is runing outside current thread, we actually need
-            # to do this to make sure the context is initialized.
-            asyncio.run_coroutine_threadsafe(_async_init(), loop=loop)
-        else:
-            async_init_done = asyncio.get_event_loop().create_task(
-                _async_init())
-            # Block until the async init finishes.
-            async_init_done.done()
-    else:
-        asyncio.get_event_loop().run_until_complete(_async_init())
-
-
 def as_future(object_id):
     """Turn an object_id into a Future object.

{"golden_diff": "diff --git a/python/ray/experimental/async_api.py b/python/ray/experimental/async_api.py\n--- a/python/ray/experimental/async_api.py\n+++ b/python/ray/experimental/async_api.py\n@@ -1,7 +1,4 @@\n-# Note: asyncio is only compatible with Python 3\n-\n import asyncio\n-import threading\n \n import ray\n from ray.experimental.async_plasma import PlasmaEventHandler\n@@ -10,7 +7,10 @@\n handler = None\n \n \n-async def _async_init():\n+def init():\n+    \"\"\"Initialize plasma event handlers for asyncio support.\"\"\"\n+    assert ray.is_initialized(), \"Please call ray.init before async_api.init\"\n+\n     global handler\n     if handler is None:\n         worker = ray.worker.global_worker\n@@ -20,31 +20,6 @@\n         logger.debug(\"AsyncPlasma Connection Created!\")\n \n \n-def init():\n-    \"\"\"\n-    Initialize synchronously.\n-    \"\"\"\n-    assert ray.is_initialized(), \"Please call ray.init before async_api.init\"\n-\n-    # Noop when handler is set.\n-    if handler is not None:\n-        return\n-\n-    loop = asyncio.get_event_loop()\n-    if loop.is_running():\n-        if loop._thread_id != threading.get_ident():\n-            # If the loop is runing outside current thread, we actually need\n-            # to do this to make sure the context is initialized.\n-            asyncio.run_coroutine_threadsafe(_async_init(), loop=loop)\n-        else:\n-            async_init_done = asyncio.get_event_loop().create_task(\n-                _async_init())\n-            # Block until the async init finishes.\n-            async_init_done.done()\n-    else:\n-        asyncio.get_event_loop().run_until_complete(_async_init())\n-\n-\n def as_future(object_id):\n     \"\"\"Turn an object_id into a Future object.\n", "issue": "Ray async api is not working with uvloop.\n<!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->\r\n\r\n### What is the problem?\r\n\r\nCurrent Ray async api uses asyncio event loop's internal attribute to identify if the loop is running in the current current thread.\r\n\r\n```python3\r\n    loop = asyncio.get_event_loop()\r\n    if loop.is_running():\r\n        if loop._thread_id != threading.get_ident():\r\n            # If the loop is runing outside current thread, we actually need\r\n            # to do this to make sure the context is initialized.\r\n            asyncio.run_coroutine_threadsafe(_async_init(), loop=loop)\r\n```\r\nThis causes a problem when we uses Ray APIs inside Fast API because Fast API uses uvloop as its main event loop, and uvloop doesn't have `_thread_id` attribute.\r\n\r\n@simon-mo Any good idea to fix this? It doesn't seem to be trivial. What about we do async_init() whenever asyncio loop is created in a different thread instead of checking if the event loop's thread id? I assume the only use case where asyncio loop is defined in a different thread is only inside async actor? \r\n\r\n### Reproduction (REQUIRED)\r\nPlease provide a script that can be run to reproduce the issue. The script should have **no external library dependencies** (i.e., use fake or mock data / environments):\r\n\r\n```python3\r\nimport time\r\nimport asyncio \r\n\u200b\r\nimport ray\r\nimport psutil\r\nfrom fastapi import FastAPI, APIRouter\r\n\u200b\r\n\u200b\r\napp = FastAPI(\r\n    title=\"API template\",\r\n    description=\"Template to build upon for API serving and distributed computation\",\r\n    version=\"0.1.0\",\r\n    openapi_url=\"/openapi.json\",\r\n    docs_url=\"/docs\",\r\n)\r\n\u200b\r\n@app.on_event(\"startup\")\r\ndef startup_event():\r\n    ray.init(num_cpus=2)\r\n\u200b\r\n\u200b\r\n\u200b\r\n@app.on_event(\"shutdown\")\r\ndef shutdown_event():\r\n    ray.shutdown()\r\n\u200b\r\n\u200b\r\n@app.get('/async')\r\nasync def non_seq_async_process():\r\n    \"\"\"\r\n    async distributed execution\r\n    \"\"\"\r\n    @ray.remote\r\n    def slow_function(i):\r\n        time.sleep(i)\r\n        return i\r\n\u200b\r\n    start_time = time.time()\r\n\u200b\r\n    # result_ids = []\r\n    # for i in range(10, 60, 10):\r\n    #     result_ids.append(slow_function.remote(i))\r\n    \r\n    # results = ray.get(result_ids)\r\n\u200b\r\n    results = await asyncio.wait([slow_function.remote(i) for i in range(10, 60, 10)])\r\n\u200b\r\n    \r\n    duration = time.time() - start_time\r\n    out = \"Executing the for loop took {:.3f} seconds.\\n\".format(duration)\r\n    out += f\"The results are: {results}\\n\"\r\n\r\n```\r\n\r\nIf we cannot run your script, we cannot fix your issue.\r\n\r\n- [x] I have verified my script runs in a clean environment and reproduces the issue.\r\n- [x] I have verified the issue also occurs with the [latest wheels](https://docs.ray.io/en/latest/installation.html).\r\n\n", "before_files": [{"content": "# Note: asyncio is only compatible with Python 3\n\nimport asyncio\nimport threading\n\nimport ray\nfrom ray.experimental.async_plasma import PlasmaEventHandler\nfrom ray.services import logger\n\nhandler = None\n\n\nasync def _async_init():\n    global handler\n    if handler is None:\n        worker = ray.worker.global_worker\n        loop = asyncio.get_event_loop()\n        handler = PlasmaEventHandler(loop, worker)\n        worker.core_worker.set_plasma_added_callback(handler)\n        logger.debug(\"AsyncPlasma Connection Created!\")\n\n\ndef init():\n    \"\"\"\n    Initialize synchronously.\n    \"\"\"\n    assert ray.is_initialized(), \"Please call ray.init before async_api.init\"\n\n    # Noop when handler is set.\n    if handler is not None:\n        return\n\n    loop = asyncio.get_event_loop()\n    if loop.is_running():\n        if loop._thread_id != threading.get_ident():\n            # If the loop is runing outside current thread, we actually need\n            # to do this to make sure the context is initialized.\n            asyncio.run_coroutine_threadsafe(_async_init(), loop=loop)\n        else:\n            async_init_done = asyncio.get_event_loop().create_task(\n                _async_init())\n            # Block until the async init finishes.\n            async_init_done.done()\n    else:\n        asyncio.get_event_loop().run_until_complete(_async_init())\n\n\ndef as_future(object_id):\n    \"\"\"Turn an object_id into a Future object.\n\n    Args:\n        object_id: A Ray object_id.\n\n    Returns:\n        PlasmaObjectFuture: A future object that waits the object_id.\n    \"\"\"\n    if handler is None:\n        init()\n    return handler.as_future(object_id)\n\n\ndef shutdown():\n    \"\"\"Manually shutdown the async API.\n\n    Cancels all related tasks and all the socket transportation.\n    \"\"\"\n    global handler\n    if handler is not None:\n        handler.close()\n        handler = None\n", "path": "python/ray/experimental/async_api.py"}], "after_files": [{"content": "import asyncio\n\nimport ray\nfrom ray.experimental.async_plasma import PlasmaEventHandler\nfrom ray.services import logger\n\nhandler = None\n\n\ndef init():\n    \"\"\"Initialize plasma event handlers for asyncio support.\"\"\"\n    assert ray.is_initialized(), \"Please call ray.init before async_api.init\"\n\n    global handler\n    if handler is None:\n        worker = ray.worker.global_worker\n        loop = asyncio.get_event_loop()\n        handler = PlasmaEventHandler(loop, worker)\n        worker.core_worker.set_plasma_added_callback(handler)\n        logger.debug(\"AsyncPlasma Connection Created!\")\n\n\ndef as_future(object_id):\n    \"\"\"Turn an object_id into a Future object.\n\n    Args:\n        object_id: A Ray object_id.\n\n    Returns:\n        PlasmaObjectFuture: A future object that waits the object_id.\n    \"\"\"\n    if handler is None:\n        init()\n    return handler.as_future(object_id)\n\n\ndef shutdown():\n    \"\"\"Manually shutdown the async API.\n\n    Cancels all related tasks and all the socket transportation.\n    \"\"\"\n    global handler\n    if handler is not None:\n        handler.close()\n        handler = None\n", "path": "python/ray/experimental/async_api.py"}]}