prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  [Asyncio] InvalidStateError when multiple awaits on same oid
  <!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->
  
  ### What is the problem?
  
  *Ray version and other system information (Python version, TensorFlow version, OS):*
  
  ### 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):
  
  If we cannot run your script, we cannot fix your issue.
  ```python
  import ray
  import time
  
  ray.init()
  @ray.remote
  def f():
      time.sleep(5)
  
  oid = f.remote()
  await asyncio.wait_for(oid, timeout=1)
  await asyncio.wait_for(oid, timeout=1)
  ```
  
  Output
  ```
  Exception in callback get_async.<locals>.done_callback(<Future finis... result=None)>) at /Users/simonmo/Desktop/ray/ray/python/ray/async_compat.py:65
  handle: <Handle get_async.<locals>.done_callback(<Future finis... result=None)>) at /Users/simonmo/Desktop/ray/ray/python/ray/async_compat.py:65>
  Traceback (most recent call last):
    File "/Users/simonmo/miniconda3/lib/python3.6/asyncio/events.py", line 145, in _run
      self._callback(*self._args)
    File "/Users/simonmo/Desktop/ray/ray/python/ray/async_compat.py", line 83, in done_callback
      user_future.set_result(result.result)
  asyncio.base_futures.InvalidStateError: invalid state
  ```
  - [ ] I have verified my script runs in a clean environment and reproduces the issue.
  - [ ] 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/async_compat.py]
  1 """
  2 This file should only be imported from Python 3.
  3 It will raise SyntaxError when importing from Python 2.
  4 """
  5 import asyncio
  6 from collections import namedtuple
  7 import time
  8 import inspect
  9 
  10 import ray
  11 
  12 
  13 def sync_to_async(func):
  14     """Convert a blocking function to async function"""
  15 
  16     if inspect.iscoroutinefunction(func):
  17         return func
  18 
  19     async def wrapper(*args, **kwargs):
  20         return func(*args, **kwargs)
  21 
  22     return wrapper
  23 
  24 
  25 # Class encapsulate the get result from direct actor.
  26 # Case 1: plasma_fallback_id=None, result=<Object>
  27 # Case 2: plasma_fallback_id=ObjectID, result=None
  28 AsyncGetResponse = namedtuple("AsyncGetResponse",
  29                               ["plasma_fallback_id", "result"])
  30 
  31 
  32 def get_async(object_id):
  33     """Asyncio compatible version of ray.get"""
  34     # Delayed import because raylet import this file and
  35     # it creates circular imports.
  36     from ray.experimental.async_api import init as async_api_init, as_future
  37     from ray.experimental.async_plasma import PlasmaObjectFuture
  38 
  39     assert isinstance(object_id, ray.ObjectID), "Batched get is not supported."
  40 
  41     # Setup
  42     async_api_init()
  43     loop = asyncio.get_event_loop()
  44     core_worker = ray.worker.global_worker.core_worker
  45 
  46     # Here's the callback used to implement async get logic.
  47     # What we want:
  48     # - If direct call, first try to get it from in memory store.
  49     #   If the object if promoted to plasma, retry it from plasma API.
  50     # - If not direct call, directly use plasma API to get it.
  51     user_future = loop.create_future()
  52 
  53     # We have three future objects here.
  54     # user_future is directly returned to the user from this function.
  55     #     and it will be eventually fulfilled by the final result.
  56     # inner_future is the first attempt to retrieve the object. It can be
  57     #     fulfilled by either core_worker.get_async or plasma_api.as_future.
  58     #     When inner_future completes, done_callback will be invoked. This
  59     #     callback set the final object in user_future if the object hasn't
  60     #     been promoted by plasma, otherwise it will retry from plasma.
  61     # retry_plasma_future is only created when we are getting objects that's
  62     #     promoted to plasma. It will also invoke the done_callback when it's
  63     #     fulfilled.
  64 
  65     def done_callback(future):
  66         result = future.result()
  67         # Result from async plasma, transparently pass it to user future
  68         if isinstance(future, PlasmaObjectFuture):
  69             if isinstance(result, ray.exceptions.RayTaskError):
  70                 ray.worker.last_task_error_raise_time = time.time()
  71                 user_future.set_exception(result.as_instanceof_cause())
  72             else:
  73                 user_future.set_result(result)
  74         else:
  75             # Result from direct call.
  76             assert isinstance(result, AsyncGetResponse), result
  77             if result.plasma_fallback_id is None:
  78                 if isinstance(result.result, ray.exceptions.RayTaskError):
  79                     ray.worker.last_task_error_raise_time = time.time()
  80                     user_future.set_exception(
  81                         result.result.as_instanceof_cause())
  82                 else:
  83                     user_future.set_result(result.result)
  84             else:
  85                 # Schedule plasma to async get, use the the same callback.
  86                 retry_plasma_future = as_future(result.plasma_fallback_id)
  87                 retry_plasma_future.add_done_callback(done_callback)
  88                 # A hack to keep reference to the future so it doesn't get GC.
  89                 user_future.retry_plasma_future = retry_plasma_future
  90 
  91     inner_future = loop.create_future()
  92     # We must add the done_callback before sending to in_memory_store_get
  93     inner_future.add_done_callback(done_callback)
  94     core_worker.in_memory_store_get_async(object_id, inner_future)
  95     # A hack to keep reference to inner_future so it doesn't get GC.
  96     user_future.inner_future = inner_future
  97     # A hack to keep a reference to the object ID for ref counting.
  98     user_future.object_id = object_id
  99 
  100     return user_future
  101
  [end of python/ray/async_compat.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/async_compat.py b/python/ray/async_compat.py
  --- a/python/ray/async_compat.py
  +++ b/python/ray/async_compat.py
  @@ -75,6 +75,11 @@
               # Result from direct call.
               assert isinstance(result, AsyncGetResponse), result
               if result.plasma_fallback_id is None:
  +                # If this future has result set already, we just need to
  +                # skip the set result/exception procedure.
  +                if user_future.done():
  +                    return
  +
                   if isinstance(result.result, ray.exceptions.RayTaskError):
                       ray.worker.last_task_error_raise_time = time.time()
                       user_future.set_exception(
  
verification_info: {"golden_diff": "diff --git a/python/ray/async_compat.py b/python/ray/async_compat.py\n--- a/python/ray/async_compat.py\n+++ b/python/ray/async_compat.py\n@@ -75,6 +75,11 @@\n             # Result from direct call.\n             assert isinstance(result, AsyncGetResponse), result\n             if result.plasma_fallback_id is None:\n+                # If this future has result set already, we just need to\n+                # skip the set result/exception procedure.\n+                if user_future.done():\n+                    return\n+\n                 if isinstance(result.result, ray.exceptions.RayTaskError):\n                     ray.worker.last_task_error_raise_time = time.time()\n                     user_future.set_exception(\n", "issue": "[Asyncio] InvalidStateError when multiple awaits on same oid\n<!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->\r\n\r\n### What is the problem?\r\n\r\n*Ray version and other system information (Python version, TensorFlow version, OS):*\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\nIf we cannot run your script, we cannot fix your issue.\r\n```python\r\nimport ray\r\nimport time\r\n\r\nray.init()\r\n@ray.remote\r\ndef f():\r\n    time.sleep(5)\r\n\r\noid = f.remote()\r\nawait asyncio.wait_for(oid, timeout=1)\r\nawait asyncio.wait_for(oid, timeout=1)\r\n```\r\n\r\nOutput\r\n```\r\nException in callback get_async.<locals>.done_callback(<Future finis... result=None)>) at /Users/simonmo/Desktop/ray/ray/python/ray/async_compat.py:65\r\nhandle: <Handle get_async.<locals>.done_callback(<Future finis... result=None)>) at /Users/simonmo/Desktop/ray/ray/python/ray/async_compat.py:65>\r\nTraceback (most recent call last):\r\n  File \"/Users/simonmo/miniconda3/lib/python3.6/asyncio/events.py\", line 145, in _run\r\n    self._callback(*self._args)\r\n  File \"/Users/simonmo/Desktop/ray/ray/python/ray/async_compat.py\", line 83, in done_callback\r\n    user_future.set_result(result.result)\r\nasyncio.base_futures.InvalidStateError: invalid state\r\n```\r\n- [ ] I have verified my script runs in a clean environment and reproduces the issue.\r\n- [ ] 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": "\"\"\"\nThis file should only be imported from Python 3.\nIt will raise SyntaxError when importing from Python 2.\n\"\"\"\nimport asyncio\nfrom collections import namedtuple\nimport time\nimport inspect\n\nimport ray\n\n\ndef sync_to_async(func):\n    \"\"\"Convert a blocking function to async function\"\"\"\n\n    if inspect.iscoroutinefunction(func):\n        return func\n\n    async def wrapper(*args, **kwargs):\n        return func(*args, **kwargs)\n\n    return wrapper\n\n\n# Class encapsulate the get result from direct actor.\n# Case 1: plasma_fallback_id=None, result=<Object>\n# Case 2: plasma_fallback_id=ObjectID, result=None\nAsyncGetResponse = namedtuple(\"AsyncGetResponse\",\n                              [\"plasma_fallback_id\", \"result\"])\n\n\ndef get_async(object_id):\n    \"\"\"Asyncio compatible version of ray.get\"\"\"\n    # Delayed import because raylet import this file and\n    # it creates circular imports.\n    from ray.experimental.async_api import init as async_api_init, as_future\n    from ray.experimental.async_plasma import PlasmaObjectFuture\n\n    assert isinstance(object_id, ray.ObjectID), \"Batched get is not supported.\"\n\n    # Setup\n    async_api_init()\n    loop = asyncio.get_event_loop()\n    core_worker = ray.worker.global_worker.core_worker\n\n    # Here's the callback used to implement async get logic.\n    # What we want:\n    # - If direct call, first try to get it from in memory store.\n    #   If the object if promoted to plasma, retry it from plasma API.\n    # - If not direct call, directly use plasma API to get it.\n    user_future = loop.create_future()\n\n    # We have three future objects here.\n    # user_future is directly returned to the user from this function.\n    #     and it will be eventually fulfilled by the final result.\n    # inner_future is the first attempt to retrieve the object. It can be\n    #     fulfilled by either core_worker.get_async or plasma_api.as_future.\n    #     When inner_future completes, done_callback will be invoked. This\n    #     callback set the final object in user_future if the object hasn't\n    #     been promoted by plasma, otherwise it will retry from plasma.\n    # retry_plasma_future is only created when we are getting objects that's\n    #     promoted to plasma. It will also invoke the done_callback when it's\n    #     fulfilled.\n\n    def done_callback(future):\n        result = future.result()\n        # Result from async plasma, transparently pass it to user future\n        if isinstance(future, PlasmaObjectFuture):\n            if isinstance(result, ray.exceptions.RayTaskError):\n                ray.worker.last_task_error_raise_time = time.time()\n                user_future.set_exception(result.as_instanceof_cause())\n            else:\n                user_future.set_result(result)\n        else:\n            # Result from direct call.\n            assert isinstance(result, AsyncGetResponse), result\n            if result.plasma_fallback_id is None:\n                if isinstance(result.result, ray.exceptions.RayTaskError):\n                    ray.worker.last_task_error_raise_time = time.time()\n                    user_future.set_exception(\n                        result.result.as_instanceof_cause())\n                else:\n                    user_future.set_result(result.result)\n            else:\n                # Schedule plasma to async get, use the the same callback.\n                retry_plasma_future = as_future(result.plasma_fallback_id)\n                retry_plasma_future.add_done_callback(done_callback)\n                # A hack to keep reference to the future so it doesn't get GC.\n                user_future.retry_plasma_future = retry_plasma_future\n\n    inner_future = loop.create_future()\n    # We must add the done_callback before sending to in_memory_store_get\n    inner_future.add_done_callback(done_callback)\n    core_worker.in_memory_store_get_async(object_id, inner_future)\n    # A hack to keep reference to inner_future so it doesn't get GC.\n    user_future.inner_future = inner_future\n    # A hack to keep a reference to the object ID for ref counting.\n    user_future.object_id = object_id\n\n    return user_future\n", "path": "python/ray/async_compat.py"}]}