We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] ActorCaller.call is at risk got hang!
<!--
Thank you for your contribution!

Please review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.
-->

**Describe the bug**
ActorCaller._listen is at risk and will not remake related futures after the client closed. It will happen when the client.recv got successfully returned with the client closed.

**To Reproduce**
To help us reproducing this bug, please provide information below:
1. Your Python version
2. The version of Mars you use
3. Versions of crucial packages, such as numpy, scipy and pandas
4. Full stack of the error.
5. Minimized code to reproduce the error.

code:
```python
import asyncio
import pytest
import mock

from ..backends.router import Router
from ..backends.core import ActorCaller
from ..errors import ServerClosed

@pytest.mark.asyncio
@mock.patch.object(Router, "get_client")
async def test_send_when_close(fake_get_client):

    class FakeClient:
        def __init__(self):
            self.closed = False
            self.send_num = 0
            self._messages = asyncio.Queue()
            self.dest_address = "test"

        async def send(self, message):
            await self._messages.put(message)
            self.send_num += 1
            if self.send_num >= 3:
                raise ConnectionError("test")

        async def recv(self, *args, **kwargs):
            await asyncio.sleep(3)
            res = await self._messages.get()
            return res

        async def close(self):
            self.closed = True

    fake_client = FakeClient()
    fake_get_client.side_effect = lambda *args, **kwargs: fake_client

    class FakeMessage:
        def __init__(self, id_num):
            self.message_id = id_num

    caller = ActorCaller()

    router = Router(
        external_addresses = ["test1"],
        local_address = "test2",
    )
    futures = []
    for index in range(2):
        futures.append(await caller.call(
            router = router,
            dest_address = "test1",
            message = FakeMessage(index),
            wait = False))

    with pytest.raises(ServerClosed):
        # Just wait _list run.
        await asyncio.sleep(1)
        await caller.call(
            router = router,
            dest_address = "test1",
            message = FakeMessage(2),
            wait = False)

    res0 = await futures[0]
    assert res0.message_id == 0

    with pytest.raises(ServerClosed):
        # This code will hand in here.
        await futures[1] 
```

**Expected behavior**
A clear and concise description of what you expected to happen.

**Additional context**
Add any other context about the problem here.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ci/importcheck.py`
Content:
```
1 #!/usr/bin/env python
2 import ast
3 import os
4 import sys
5 import textwrap
6 from pathlib import PurePath
7 from typing import List, NamedTuple, Optional, Tuple
8 
9 _IGNORES = [
10     'mars/lib/**/*.py',
11     'conftest.py',
12 ]
13 
14 
15 class CheckResult(NamedTuple):
16     path: str
17     lines: List
18     absolute_imports: List[Tuple[int, int]]
19     head_disorder: Optional[Tuple[int, int]]
20     block_disorders: Optional[List[Tuple[int, int]]]
21 
22     @property
23     def has_faults(self) -> bool:
24         return bool(self.absolute_imports) or bool(self.head_disorder) \
25             or bool(self.block_disorders)
26 
27 
28 def _check_absolute_import(node: ast.AST) -> List[Tuple[int, int]]:
29     res = set()
30     if isinstance(node, ast.Import):
31         for import_name in node.names:
32             if import_name.name.startswith('mars.'):
33                 res.add((node.lineno, node.end_lineno))
34     elif isinstance(node, ast.ImportFrom):
35         if node.level == 0 and node.module.startswith('mars.'):
36             res.add((node.lineno, node.end_lineno))
37     elif getattr(node, 'body', []):
38         for body_item in node.body:
39             res.update(_check_absolute_import(body_item))
40     return sorted(res)
41 
42 
43 def check_imports(file_path) -> CheckResult:
44     with open(file_path, 'rb') as src_file:
45         body = src_file.read()
46         lines = body.splitlines()
47         parsed = ast.parse(body, filename=file_path)
48     # scan for imports
49     abs_faults = _check_absolute_import(parsed)
50 
51     return CheckResult(file_path, lines, abs_faults, None, None)
52 
53 
54 def _extract_line_block(lines: List, lineno: int, end_lineno: int, indent: str):
55     grab_lines = '\n'.join(line.decode() for line in lines[lineno - 1:end_lineno])
56     return textwrap.indent(textwrap.dedent(grab_lines), indent)
57 
58 
59 def format_results(results: List[CheckResult], root_path):
60     rel_import_count = sum(len(res.absolute_imports) for res in results)
61     if rel_import_count > 0:
62         print(f'Do not use absolute imports for mars module in non-test '
63               f'code ({rel_import_count}):', file=sys.stderr)
64         for res in results:
65             if not res.absolute_imports:
66                 continue
67             rel_path = os.path.relpath(res.path, root_path)
68             print('  ' + rel_path, file=sys.stderr)
69             for lineno, end_lineno in res.absolute_imports:
70                 print(f'    Line {lineno}-{end_lineno}', file=sys.stderr)
71                 print(_extract_line_block(res.lines, lineno, end_lineno, '      '),
72                       file=sys.stderr)
73 
74 
75 def main():
76     root_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
77     results = []
78     for root, _dirs, files in os.walk(os.path.join(root_path, 'mars')):
79         if 'tests' in root:
80             continue
81         for fn in files:
82             abs_path = os.path.join(root, fn)
83             rel_path = os.path.relpath(abs_path, root_path)
84 
85             if not fn.endswith('.py'):
86                 continue
87             if any(PurePath(rel_path).match(patt) for patt in _IGNORES):
88                 continue
89 
90             check_result = check_imports(abs_path)
91             if check_result.has_faults:
92                 results.append(check_result)
93     if results:
94         results = sorted(results, key=lambda x: x.path)
95         format_results(results, root_path)
96         sys.exit(1)
97 
98 
99 if __name__ == '__main__':
100     main()
101
```

Path: `mars/oscar/backends/core.py`
Content:
```
1 # Copyright 1999-2021 Alibaba Group Holding Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import asyncio
16 from typing import Dict, Union
17 
18 from ..errors import ServerClosed
19 from .communication import Client
20 from .message import _MessageBase, ResultMessage, ErrorMessage, \
21     DeserializeMessageFailed
22 from .router import Router
23 
24 
25 result_message_type = Union[ResultMessage, ErrorMessage]
26 
27 
28 class ActorCaller:
29     __slots__ = '_client_to_message_futures', '_clients'
30 
31     def __init__(self):
32         self._client_to_message_futures: \
33             Dict[Client, Dict[bytes, asyncio.Future]] = dict()
34         self._clients: Dict[Client, asyncio.Task] = dict()
35 
36     async def get_client(self,
37                          router: Router,
38                          dest_address: str) -> Client:
39         client = await router.get_client(dest_address,
40                                          from_who=self)
41         if client not in self._clients:
42             self._clients[client] = asyncio.create_task(self._listen(client))
43             self._client_to_message_futures[client] = dict()
44         return client
45 
46     async def _listen(self, client: Client):
47         while not client.closed:
48             try:
49                 try:
50                     message: _MessageBase = await client.recv()
51                 except (EOFError, ConnectionError, BrokenPipeError):
52                     # remote server closed, close client and raise ServerClosed
53                     try:
54                         await client.close()
55                     except (ConnectionError, BrokenPipeError):
56                         # close failed, ignore it
57                         pass
58                     raise ServerClosed(f'Remote server {client.dest_address} closed') from None
59                 future = self._client_to_message_futures[client].pop(message.message_id)
60                 future.set_result(message)
61             except DeserializeMessageFailed as e:  # pragma: no cover
62                 message_id = e.message_id
63                 future = self._client_to_message_futures[client].pop(message_id)
64                 future.set_exception(e)
65             except Exception as e:  # noqa: E722  # pylint: disable=bare-except
66                 message_futures = self._client_to_message_futures.get(client)
67                 self._client_to_message_futures[client] = dict()
68                 for future in message_futures.values():
69                     future.set_exception(e)
70 
71     async def call(self,
72                    router: Router,
73                    dest_address: str,
74                    message: _MessageBase,
75                    wait: bool = True) \
76             -> Union[ResultMessage, ErrorMessage, asyncio.Future]:
77         client = await self.get_client(router, dest_address)
78         loop = asyncio.get_running_loop()
79         wait_response = loop.create_future()
80         self._client_to_message_futures[client][message.message_id] = wait_response
81 
82         try:
83             await client.send(message)
84         except ConnectionError:
85             try:
86                 await client.close()
87             except ConnectionError:
88                 # close failed, ignore it
89                 pass
90             raise ServerClosed(f'Remote server {client.dest_address} closed')
91 
92         if not wait:
93             return wait_response
94         else:
95             return await wait_response
96 
97     async def stop(self):
98         try:
99             await asyncio.gather(*[client.close() for client in self._clients])
100         except (ConnectionError, ServerClosed):
101             pass
102         self.cancel_tasks()
103 
104     def cancel_tasks(self):
105         # cancel listening for all clients
106         _ = [task.cancel() for task in self._clients.values()]
107
```
--- 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/ci/importcheck.py b/ci/importcheck.py
--- a/ci/importcheck.py
+++ b/ci/importcheck.py
@@ -1,4 +1,18 @@
 #!/usr/bin/env python
+# Copyright 1999-2021 Alibaba Group Holding Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import ast
 import os
 import sys
diff --git a/mars/oscar/backends/core.py b/mars/oscar/backends/core.py
--- a/mars/oscar/backends/core.py
+++ b/mars/oscar/backends/core.py
@@ -68,6 +68,12 @@
                 for future in message_futures.values():
                     future.set_exception(e)
 
+        message_futures = self._client_to_message_futures.get(client)
+        self._client_to_message_futures[client] = dict()
+        error = ServerClosed(f'Remote server {client.dest_address} closed')
+        for future in message_futures.values():
+            future.set_exception(error)
+
     async def call(self,
                    router: Router,
                    dest_address: str,

{"golden_diff": "diff --git a/ci/importcheck.py b/ci/importcheck.py\n--- a/ci/importcheck.py\n+++ b/ci/importcheck.py\n@@ -1,4 +1,18 @@\n #!/usr/bin/env python\n+# Copyright 1999-2021 Alibaba Group Holding Ltd.\n+#\n+# Licensed under the Apache License, Version 2.0 (the \"License\");\n+# you may not use this file except in compliance with the License.\n+# You may obtain a copy of the License at\n+#\n+#      http://www.apache.org/licenses/LICENSE-2.0\n+#\n+# Unless required by applicable law or agreed to in writing, software\n+# distributed under the License is distributed on an \"AS IS\" BASIS,\n+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+# See the License for the specific language governing permissions and\n+# limitations under the License.\n+\n import ast\n import os\n import sys\ndiff --git a/mars/oscar/backends/core.py b/mars/oscar/backends/core.py\n--- a/mars/oscar/backends/core.py\n+++ b/mars/oscar/backends/core.py\n@@ -68,6 +68,12 @@\n                 for future in message_futures.values():\n                     future.set_exception(e)\n \n+        message_futures = self._client_to_message_futures.get(client)\n+        self._client_to_message_futures[client] = dict()\n+        error = ServerClosed(f'Remote server {client.dest_address} closed')\n+        for future in message_futures.values():\n+            future.set_exception(error)\n+\n     async def call(self,\n                    router: Router,\n                    dest_address: str,\n", "issue": "[BUG] ActorCaller.call is at risk got hang!\n<!--\r\nThank you for your contribution!\r\n\r\nPlease review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.\r\n-->\r\n\r\n**Describe the bug**\r\nActorCaller._listen is at risk and will not remake related futures after the client closed. It will happen when the client.recv got successfully returned with the client closed.\r\n\r\n**To Reproduce**\r\nTo help us reproducing this bug, please provide information below:\r\n1. Your Python version\r\n2. The version of Mars you use\r\n3. Versions of crucial packages, such as numpy, scipy and pandas\r\n4. Full stack of the error.\r\n5. Minimized code to reproduce the error.\r\n\r\ncode:\r\n```python\r\nimport asyncio\r\nimport pytest\r\nimport mock\r\n\r\nfrom ..backends.router import Router\r\nfrom ..backends.core import ActorCaller\r\nfrom ..errors import ServerClosed\r\n\r\n@pytest.mark.asyncio\r\n@mock.patch.object(Router, \"get_client\")\r\nasync def test_send_when_close(fake_get_client):\r\n\r\n    class FakeClient:\r\n        def __init__(self):\r\n            self.closed = False\r\n            self.send_num = 0\r\n            self._messages = asyncio.Queue()\r\n            self.dest_address = \"test\"\r\n\r\n        async def send(self, message):\r\n            await self._messages.put(message)\r\n            self.send_num += 1\r\n            if self.send_num >= 3:\r\n                raise ConnectionError(\"test\")\r\n\r\n        async def recv(self, *args, **kwargs):\r\n            await asyncio.sleep(3)\r\n            res = await self._messages.get()\r\n            return res\r\n\r\n        async def close(self):\r\n            self.closed = True\r\n\r\n    fake_client = FakeClient()\r\n    fake_get_client.side_effect = lambda *args, **kwargs: fake_client\r\n\r\n    class FakeMessage:\r\n        def __init__(self, id_num):\r\n            self.message_id = id_num\r\n\r\n    caller = ActorCaller()\r\n\r\n    router = Router(\r\n        external_addresses = [\"test1\"],\r\n        local_address = \"test2\",\r\n    )\r\n    futures = []\r\n    for index in range(2):\r\n        futures.append(await caller.call(\r\n            router = router,\r\n            dest_address = \"test1\",\r\n            message = FakeMessage(index),\r\n            wait = False))\r\n\r\n    with pytest.raises(ServerClosed):\r\n        # Just wait _list run.\r\n        await asyncio.sleep(1)\r\n        await caller.call(\r\n            router = router,\r\n            dest_address = \"test1\",\r\n            message = FakeMessage(2),\r\n            wait = False)\r\n\r\n    res0 = await futures[0]\r\n    assert res0.message_id == 0\r\n\r\n    with pytest.raises(ServerClosed):\r\n        # This code will hand in here.\r\n        await futures[1] \r\n```\r\n\r\n**Expected behavior**\r\nA clear and concise description of what you expected to happen.\r\n\r\n**Additional context**\r\nAdd any other context about the problem here.\r\n\n", "before_files": [{"content": "#!/usr/bin/env python\nimport ast\nimport os\nimport sys\nimport textwrap\nfrom pathlib import PurePath\nfrom typing import List, NamedTuple, Optional, Tuple\n\n_IGNORES = [\n    'mars/lib/**/*.py',\n    'conftest.py',\n]\n\n\nclass CheckResult(NamedTuple):\n    path: str\n    lines: List\n    absolute_imports: List[Tuple[int, int]]\n    head_disorder: Optional[Tuple[int, int]]\n    block_disorders: Optional[List[Tuple[int, int]]]\n\n    @property\n    def has_faults(self) -> bool:\n        return bool(self.absolute_imports) or bool(self.head_disorder) \\\n            or bool(self.block_disorders)\n\n\ndef _check_absolute_import(node: ast.AST) -> List[Tuple[int, int]]:\n    res = set()\n    if isinstance(node, ast.Import):\n        for import_name in node.names:\n            if import_name.name.startswith('mars.'):\n                res.add((node.lineno, node.end_lineno))\n    elif isinstance(node, ast.ImportFrom):\n        if node.level == 0 and node.module.startswith('mars.'):\n            res.add((node.lineno, node.end_lineno))\n    elif getattr(node, 'body', []):\n        for body_item in node.body:\n            res.update(_check_absolute_import(body_item))\n    return sorted(res)\n\n\ndef check_imports(file_path) -> CheckResult:\n    with open(file_path, 'rb') as src_file:\n        body = src_file.read()\n        lines = body.splitlines()\n        parsed = ast.parse(body, filename=file_path)\n    # scan for imports\n    abs_faults = _check_absolute_import(parsed)\n\n    return CheckResult(file_path, lines, abs_faults, None, None)\n\n\ndef _extract_line_block(lines: List, lineno: int, end_lineno: int, indent: str):\n    grab_lines = '\\n'.join(line.decode() for line in lines[lineno - 1:end_lineno])\n    return textwrap.indent(textwrap.dedent(grab_lines), indent)\n\n\ndef format_results(results: List[CheckResult], root_path):\n    rel_import_count = sum(len(res.absolute_imports) for res in results)\n    if rel_import_count > 0:\n        print(f'Do not use absolute imports for mars module in non-test '\n              f'code ({rel_import_count}):', file=sys.stderr)\n        for res in results:\n            if not res.absolute_imports:\n                continue\n            rel_path = os.path.relpath(res.path, root_path)\n            print('  ' + rel_path, file=sys.stderr)\n            for lineno, end_lineno in res.absolute_imports:\n                print(f'    Line {lineno}-{end_lineno}', file=sys.stderr)\n                print(_extract_line_block(res.lines, lineno, end_lineno, '      '),\n                      file=sys.stderr)\n\n\ndef main():\n    root_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))\n    results = []\n    for root, _dirs, files in os.walk(os.path.join(root_path, 'mars')):\n        if 'tests' in root:\n            continue\n        for fn in files:\n            abs_path = os.path.join(root, fn)\n            rel_path = os.path.relpath(abs_path, root_path)\n\n            if not fn.endswith('.py'):\n                continue\n            if any(PurePath(rel_path).match(patt) for patt in _IGNORES):\n                continue\n\n            check_result = check_imports(abs_path)\n            if check_result.has_faults:\n                results.append(check_result)\n    if results:\n        results = sorted(results, key=lambda x: x.path)\n        format_results(results, root_path)\n        sys.exit(1)\n\n\nif __name__ == '__main__':\n    main()\n", "path": "ci/importcheck.py"}, {"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport asyncio\nfrom typing import Dict, Union\n\nfrom ..errors import ServerClosed\nfrom .communication import Client\nfrom .message import _MessageBase, ResultMessage, ErrorMessage, \\\n    DeserializeMessageFailed\nfrom .router import Router\n\n\nresult_message_type = Union[ResultMessage, ErrorMessage]\n\n\nclass ActorCaller:\n    __slots__ = '_client_to_message_futures', '_clients'\n\n    def __init__(self):\n        self._client_to_message_futures: \\\n            Dict[Client, Dict[bytes, asyncio.Future]] = dict()\n        self._clients: Dict[Client, asyncio.Task] = dict()\n\n    async def get_client(self,\n                         router: Router,\n                         dest_address: str) -> Client:\n        client = await router.get_client(dest_address,\n                                         from_who=self)\n        if client not in self._clients:\n            self._clients[client] = asyncio.create_task(self._listen(client))\n            self._client_to_message_futures[client] = dict()\n        return client\n\n    async def _listen(self, client: Client):\n        while not client.closed:\n            try:\n                try:\n                    message: _MessageBase = await client.recv()\n                except (EOFError, ConnectionError, BrokenPipeError):\n                    # remote server closed, close client and raise ServerClosed\n                    try:\n                        await client.close()\n                    except (ConnectionError, BrokenPipeError):\n                        # close failed, ignore it\n                        pass\n                    raise ServerClosed(f'Remote server {client.dest_address} closed') from None\n                future = self._client_to_message_futures[client].pop(message.message_id)\n                future.set_result(message)\n            except DeserializeMessageFailed as e:  # pragma: no cover\n                message_id = e.message_id\n                future = self._client_to_message_futures[client].pop(message_id)\n                future.set_exception(e)\n            except Exception as e:  # noqa: E722  # pylint: disable=bare-except\n                message_futures = self._client_to_message_futures.get(client)\n                self._client_to_message_futures[client] = dict()\n                for future in message_futures.values():\n                    future.set_exception(e)\n\n    async def call(self,\n                   router: Router,\n                   dest_address: str,\n                   message: _MessageBase,\n                   wait: bool = True) \\\n            -> Union[ResultMessage, ErrorMessage, asyncio.Future]:\n        client = await self.get_client(router, dest_address)\n        loop = asyncio.get_running_loop()\n        wait_response = loop.create_future()\n        self._client_to_message_futures[client][message.message_id] = wait_response\n\n        try:\n            await client.send(message)\n        except ConnectionError:\n            try:\n                await client.close()\n            except ConnectionError:\n                # close failed, ignore it\n                pass\n            raise ServerClosed(f'Remote server {client.dest_address} closed')\n\n        if not wait:\n            return wait_response\n        else:\n            return await wait_response\n\n    async def stop(self):\n        try:\n            await asyncio.gather(*[client.close() for client in self._clients])\n        except (ConnectionError, ServerClosed):\n            pass\n        self.cancel_tasks()\n\n    def cancel_tasks(self):\n        # cancel listening for all clients\n        _ = [task.cancel() for task in self._clients.values()]\n", "path": "mars/oscar/backends/core.py"}], "after_files": [{"content": "#!/usr/bin/env python\n# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport ast\nimport os\nimport sys\nimport textwrap\nfrom pathlib import PurePath\nfrom typing import List, NamedTuple, Optional, Tuple\n\n_IGNORES = [\n    'mars/lib/**/*.py',\n    'conftest.py',\n]\n\n\nclass CheckResult(NamedTuple):\n    path: str\n    lines: List\n    absolute_imports: List[Tuple[int, int]]\n    head_disorder: Optional[Tuple[int, int]]\n    block_disorders: Optional[List[Tuple[int, int]]]\n\n    @property\n    def has_faults(self) -> bool:\n        return bool(self.absolute_imports) or bool(self.head_disorder) \\\n            or bool(self.block_disorders)\n\n\ndef _check_absolute_import(node: ast.AST) -> List[Tuple[int, int]]:\n    res = set()\n    if isinstance(node, ast.Import):\n        for import_name in node.names:\n            if import_name.name.startswith('mars.'):\n                res.add((node.lineno, node.end_lineno))\n    elif isinstance(node, ast.ImportFrom):\n        if node.level == 0 and node.module.startswith('mars.'):\n            res.add((node.lineno, node.end_lineno))\n    elif getattr(node, 'body', []):\n        for body_item in node.body:\n            res.update(_check_absolute_import(body_item))\n    return sorted(res)\n\n\ndef check_imports(file_path) -> CheckResult:\n    with open(file_path, 'rb') as src_file:\n        body = src_file.read()\n        lines = body.splitlines()\n        parsed = ast.parse(body, filename=file_path)\n    # scan for imports\n    abs_faults = _check_absolute_import(parsed)\n\n    return CheckResult(file_path, lines, abs_faults, None, None)\n\n\ndef _extract_line_block(lines: List, lineno: int, end_lineno: int, indent: str):\n    grab_lines = '\\n'.join(line.decode() for line in lines[lineno - 1:end_lineno])\n    return textwrap.indent(textwrap.dedent(grab_lines), indent)\n\n\ndef format_results(results: List[CheckResult], root_path):\n    rel_import_count = sum(len(res.absolute_imports) for res in results)\n    if rel_import_count > 0:\n        print(f'Do not use absolute imports for mars module in non-test '\n              f'code ({rel_import_count}):', file=sys.stderr)\n        for res in results:\n            if not res.absolute_imports:\n                continue\n            rel_path = os.path.relpath(res.path, root_path)\n            print('  ' + rel_path, file=sys.stderr)\n            for lineno, end_lineno in res.absolute_imports:\n                print(f'    Line {lineno}-{end_lineno}', file=sys.stderr)\n                print(_extract_line_block(res.lines, lineno, end_lineno, '      '),\n                      file=sys.stderr)\n\n\ndef main():\n    root_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))\n    results = []\n    for root, _dirs, files in os.walk(os.path.join(root_path, 'mars')):\n        if 'tests' in root:\n            continue\n        for fn in files:\n            abs_path = os.path.join(root, fn)\n            rel_path = os.path.relpath(abs_path, root_path)\n\n            if not fn.endswith('.py'):\n                continue\n            if any(PurePath(rel_path).match(patt) for patt in _IGNORES):\n                continue\n\n            check_result = check_imports(abs_path)\n            if check_result.has_faults:\n                results.append(check_result)\n    if results:\n        results = sorted(results, key=lambda x: x.path)\n        format_results(results, root_path)\n        sys.exit(1)\n\n\nif __name__ == '__main__':\n    main()\n", "path": "ci/importcheck.py"}, {"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport asyncio\nfrom typing import Dict, Union\n\nfrom ..errors import ServerClosed\nfrom .communication import Client\nfrom .message import _MessageBase, ResultMessage, ErrorMessage, \\\n    DeserializeMessageFailed\nfrom .router import Router\n\n\nresult_message_type = Union[ResultMessage, ErrorMessage]\n\n\nclass ActorCaller:\n    __slots__ = '_client_to_message_futures', '_clients'\n\n    def __init__(self):\n        self._client_to_message_futures: \\\n            Dict[Client, Dict[bytes, asyncio.Future]] = dict()\n        self._clients: Dict[Client, asyncio.Task] = dict()\n\n    async def get_client(self,\n                         router: Router,\n                         dest_address: str) -> Client:\n        client = await router.get_client(dest_address,\n                                         from_who=self)\n        if client not in self._clients:\n            self._clients[client] = asyncio.create_task(self._listen(client))\n            self._client_to_message_futures[client] = dict()\n        return client\n\n    async def _listen(self, client: Client):\n        while not client.closed:\n            try:\n                try:\n                    message: _MessageBase = await client.recv()\n                except (EOFError, ConnectionError, BrokenPipeError):\n                    # remote server closed, close client and raise ServerClosed\n                    try:\n                        await client.close()\n                    except (ConnectionError, BrokenPipeError):\n                        # close failed, ignore it\n                        pass\n                    raise ServerClosed(f'Remote server {client.dest_address} closed') from None\n                future = self._client_to_message_futures[client].pop(message.message_id)\n                future.set_result(message)\n            except DeserializeMessageFailed as e:  # pragma: no cover\n                message_id = e.message_id\n                future = self._client_to_message_futures[client].pop(message_id)\n                future.set_exception(e)\n            except Exception as e:  # noqa: E722  # pylint: disable=bare-except\n                message_futures = self._client_to_message_futures.get(client)\n                self._client_to_message_futures[client] = dict()\n                for future in message_futures.values():\n                    future.set_exception(e)\n\n        message_futures = self._client_to_message_futures.get(client)\n        self._client_to_message_futures[client] = dict()\n        error = ServerClosed(f'Remote server {client.dest_address} closed')\n        for future in message_futures.values():\n            future.set_exception(error)\n\n    async def call(self,\n                   router: Router,\n                   dest_address: str,\n                   message: _MessageBase,\n                   wait: bool = True) \\\n            -> Union[ResultMessage, ErrorMessage, asyncio.Future]:\n        client = await self.get_client(router, dest_address)\n        loop = asyncio.get_running_loop()\n        wait_response = loop.create_future()\n        self._client_to_message_futures[client][message.message_id] = wait_response\n\n        try:\n            await client.send(message)\n        except ConnectionError:\n            try:\n                await client.close()\n            except ConnectionError:\n                # close failed, ignore it\n                pass\n            raise ServerClosed(f'Remote server {client.dest_address} closed')\n\n        if not wait:\n            return wait_response\n        else:\n            return await wait_response\n\n    async def stop(self):\n        try:\n            await asyncio.gather(*[client.close() for client in self._clients])\n        except (ConnectionError, ServerClosed):\n            pass\n        self.cancel_tasks()\n\n    def cancel_tasks(self):\n        # cancel listening for all clients\n        _ = [task.cancel() for task in self._clients.values()]\n", "path": "mars/oscar/backends/core.py"}]}