prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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).
  
  
  </issue>
  <code>
  [start of python/ray/experimental/async_api.py]
  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 of python/ray/experimental/async_api.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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.
  
verification_info: {"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"}]}