We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN 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).


--- 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/async_compat.py`
Content:
```
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 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/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(

{"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"}], "after_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 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                        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"}]}