We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] Duplicate operands execution
<!--
Thank you for your contribution!

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

**Describe the bug**

The case is `mars.core.entity.tests.test_utils.test_recursive_tile`. It generates a chunk graph with duplicate operands (the op key are same, the execution output are same).

I added some code to check the operands in stage.py:

``` python
class TaskStageProcessor:
    def __init__(
        self,
        stage_id: str,
        task: Task,
        chunk_graph: ChunkGraph,
        subtask_graph: SubtaskGraph,
        bands: List[BandType],
        tile_context: Dict[TileableType, TileableType],
        scheduling_api: SchedulingAPI,
        meta_api: MetaAPI,
    ):
        self.stage_id = stage_id
        self.task = task
        self.chunk_graph = chunk_graph
        self.subtask_graph = subtask_graph
        self._bands = bands
        self._tile_context = tile_context

        # Check for duplicate operand.
        from collections import Counter
        unique_keys = Counter(c.key for c in chunk_graph)
        if len(unique_keys) < len(chunk_graph):
            import pprint
            pprint.pprint([c for c in chunk_graph])
            pprint.pprint([k for k, v in unique_keys.items() if v > 1])
            subtask_graph.view()
```

The output is:

```python
[Chunk <op=TensorRand, key=560bdcf1ae4978a999cbe3fd620c65d7_0>,
 Chunk <op=TensorRand, key=5aed4a149f5084ce95b9212afb61f24d_0>,
 Chunk <op=TensorRand, key=6fa461793a5d0feef9905c945653653f_0>,
 Chunk <op=TensorRand, key=688bd5f2d8fe39ce3e1a8af91b53b65c_0>,
 Chunk <op=TensorSlice, key=a9cd6ffd308ed11c3fccd2753f9a4aba_0>,
 Chunk <op=PSRSShuffle, stage=reduce, key=94c1f8fa627bcd5960ad795f7c692671_0>,
 Chunk <op=TensorShuffleProxy, key=6e7794b968faabcbebdfb289789c1ab9_0>,
 Chunk <op=PSRSShuffle, stage=map, key=0ebf73717ef2479456a6351c6209a90e_0>,
 Chunk <op=PSRSShuffle, stage=map, key=eb053dc8499bc7cdc12d8382f199e096_0>,
 Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_0>,
 Chunk <op=PSRSConcatPivot, key=58ce89bb01132fd0fd757246ef654c40_0>,
 Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_1>,
 Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_1>,
 Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_0>,
 Chunk <op=TensorSlice, key=bdbeadf9202694d9c8f86b9fc8fc2db2_0>,
 Chunk <op=PSRSShuffle, stage=reduce, key=22974984dc0b50cff4d077e0413202d2_0>,
 Chunk <op=PSRSShuffle, stage=reduce, key=6c78a27cd13360463638ae139958297e_0>,
 Chunk <op=TensorShuffleProxy, key=fd341cca1ae125e359e9a2274928605f_0>,
 Chunk <op=PSRSShuffle, stage=map, key=7d9f9ea924cf86e2aba771257f46db95_0>,
 Chunk <op=PSRSShuffle, stage=map, key=45359c01dfab34dd6d552687e3e9cf86_0>,
 Chunk <op=PSRSSortRegularSample, key=cac04cc478ecb467c0f1a7bca4bf809c_0>,
 Chunk <op=PSRSConcatPivot, key=ee1f386a09653687d7f3802aaf6d8cd3_0>,
 Chunk <op=PSRSSortRegularSample, key=321f2eda3a690282eb2d75ea0aa2614e_1>,
 Chunk <op=PSRSSortRegularSample, key=cac04cc478ecb467c0f1a7bca4bf809c_1>,
 Chunk <op=PSRSSortRegularSample, key=321f2eda3a690282eb2d75ea0aa2614e_0>,
 Chunk <op=PSRSShuffle, stage=reduce, key=02d268f3091f3c95271acc314e0ef0fd_0>,
 Chunk <op=PSRSShuffle, stage=reduce, key=94c1f8fa627bcd5960ad795f7c692671_0>,
 Chunk <op=TensorShuffleProxy, key=6e7794b968faabcbebdfb289789c1ab9_0>,
 Chunk <op=PSRSShuffle, stage=map, key=0ebf73717ef2479456a6351c6209a90e_0>,
 Chunk <op=PSRSShuffle, stage=map, key=eb053dc8499bc7cdc12d8382f199e096_0>,
 Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_0>,
 Chunk <op=PSRSConcatPivot, key=58ce89bb01132fd0fd757246ef654c40_0>,
 Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_1>,
 Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_1>,
 Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_0>,
 Chunk <op=PSRSShuffle, stage=reduce, key=22974984dc0b50cff4d077e0413202d2_0>]
['94c1f8fa627bcd5960ad795f7c692671_0',
 '6e7794b968faabcbebdfb289789c1ab9_0',
 '0ebf73717ef2479456a6351c6209a90e_0',
 'eb053dc8499bc7cdc12d8382f199e096_0',
 '125e3a7f78bcb6c939668f03a3b94787_0',
 '58ce89bb01132fd0fd757246ef654c40_0',
 '98e601719051e980b4d9f3f20b20e437_1',
 '125e3a7f78bcb6c939668f03a3b94787_1',
 '98e601719051e980b4d9f3f20b20e437_0',
 '22974984dc0b50cff4d077e0413202d2_0']
```

Dump graph: [default.pdf](https://github.com/mars-project/mars/files/8645133/default.pdf)

I added check output code to the `PSRSShuffle._execute_map`:
```python
for i in range(op.n_partition):
    key = (out.key, (i,))
    value = tuple(reduce_outputs[i].ravel())
    if key in ctx:
        existing_value = ctx[key]
        np.testing.assert_array_equal(existing_value, value)
        raise Exception(f"The key already exists: {key}")
    ctx[out.key, (i,)] = tuple(reduce_outputs[i].ravel())
```

It raises an exception and the `assert_array_equal` passes.

```python
            for i in range(op.n_partition):
                key = (out.key, (i,))
                value = tuple(reduce_outputs[i].ravel())
                if key in ctx:
                    existing_value = ctx[key]
                    np.testing.assert_array_equal(existing_value, value)
>                   raise Exception(f"The key already exists: {key}")
E                   Exception: The key already exists: ('eb053dc8499bc7cdc12d8382f199e096_0', (0,))

mars/tensor/base/psrs.py:753: Exception
```

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

**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: `mars/core/entity/utils.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 from typing import List, Union, Generator
16 
17 from ...typing import TileableType, ChunkType
18 from ...utils import has_unknown_shape, calc_nsplits
19 
20 
21 def refresh_tileable_shape(tileable):
22     if has_unknown_shape(tileable):
23         # update shape
24         nsplits = calc_nsplits({c.index: c.shape for c in tileable.chunks})
25         shape = tuple(sum(ns) for ns in nsplits)
26         tileable._nsplits = nsplits
27         tileable._shape = shape
28 
29 
30 def tile(tileable, *tileables: TileableType):
31     from ..graph import (
32         TileableGraph,
33         TileableGraphBuilder,
34         ChunkGraphBuilder,
35         TileContext,
36     )
37 
38     raw_tileables = target_tileables = [tileable] + list(tileables)
39     target_tileables = [t.data if hasattr(t, "data") else t for t in target_tileables]
40 
41     tileable_graph = TileableGraph(target_tileables)
42     tileable_graph_builder = TileableGraphBuilder(tileable_graph)
43     next(tileable_graph_builder.build())
44 
45     # tile
46     tile_context = TileContext()
47     chunk_graph_builder = ChunkGraphBuilder(
48         tileable_graph, fuse_enabled=False, tile_context=tile_context
49     )
50     next(chunk_graph_builder.build())
51 
52     if len(tileables) == 0:
53         return type(tileable)(tile_context[target_tileables[0]])
54     else:
55         return [
56             type(raw_t)(tile_context[t])
57             for raw_t, t in zip(raw_tileables, target_tileables)
58         ]
59 
60 
61 def recursive_tile(
62     tileable: TileableType, *tileables: TileableType
63 ) -> Generator[
64     List[ChunkType], List[ChunkType], Union[TileableType, List[TileableType]]
65 ]:
66     from .tileables import handler
67 
68     return_list = len(tileables) > 0
69     if not return_list and isinstance(tileable, (list, tuple)):
70         return_list = True
71         raw = tileable
72         tileable = raw[0]
73         tileables = raw[1:]
74 
75     to_tile = [tileable] + list(tileables)
76     q = [t for t in to_tile if t.is_coarse()]
77     while q:
78         t = q[-1]
79         cs = [c for c in t.inputs if c.is_coarse()]
80         if cs:
81             q.extend(cs)
82             continue
83         for obj in handler.tile(t.op.outputs):
84             to_update_inputs = []
85             chunks = []
86             for inp in t.op.inputs:
87                 chunks.extend(inp.chunks)
88                 if has_unknown_shape(inp):
89                     to_update_inputs.append(inp)
90             if obj is None:
91                 yield chunks + to_update_inputs
92             else:
93                 yield obj + to_update_inputs
94         q.pop()
95 
96     if not return_list:
97         return tileable
98     else:
99         return [tileable] + list(tileables)
100
```
--- 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/mars/core/entity/utils.py b/mars/core/entity/utils.py
--- a/mars/core/entity/utils.py
+++ b/mars/core/entity/utils.py
@@ -76,21 +76,24 @@
     q = [t for t in to_tile if t.is_coarse()]
     while q:
         t = q[-1]
-        cs = [c for c in t.inputs if c.is_coarse()]
-        if cs:
-            q.extend(cs)
-            continue
-        for obj in handler.tile(t.op.outputs):
-            to_update_inputs = []
-            chunks = []
-            for inp in t.op.inputs:
-                chunks.extend(inp.chunks)
-                if has_unknown_shape(inp):
-                    to_update_inputs.append(inp)
-            if obj is None:
-                yield chunks + to_update_inputs
-            else:
-                yield obj + to_update_inputs
+        if t.is_coarse():
+            # t may be put into q repeatedly,
+            # so we check if it's tiled or not
+            cs = [c for c in t.inputs if c.is_coarse()]
+            if cs:
+                q.extend(cs)
+                continue
+            for obj in handler.tile(t.op.outputs):
+                to_update_inputs = []
+                chunks = []
+                for inp in t.op.inputs:
+                    chunks.extend(inp.chunks)
+                    if has_unknown_shape(inp):
+                        to_update_inputs.append(inp)
+                if obj is None:
+                    yield chunks + to_update_inputs
+                else:
+                    yield obj + to_update_inputs
         q.pop()
 
     if not return_list:

{"golden_diff": "diff --git a/mars/core/entity/utils.py b/mars/core/entity/utils.py\n--- a/mars/core/entity/utils.py\n+++ b/mars/core/entity/utils.py\n@@ -76,21 +76,24 @@\n     q = [t for t in to_tile if t.is_coarse()]\n     while q:\n         t = q[-1]\n-        cs = [c for c in t.inputs if c.is_coarse()]\n-        if cs:\n-            q.extend(cs)\n-            continue\n-        for obj in handler.tile(t.op.outputs):\n-            to_update_inputs = []\n-            chunks = []\n-            for inp in t.op.inputs:\n-                chunks.extend(inp.chunks)\n-                if has_unknown_shape(inp):\n-                    to_update_inputs.append(inp)\n-            if obj is None:\n-                yield chunks + to_update_inputs\n-            else:\n-                yield obj + to_update_inputs\n+        if t.is_coarse():\n+            # t may be put into q repeatedly,\n+            # so we check if it's tiled or not\n+            cs = [c for c in t.inputs if c.is_coarse()]\n+            if cs:\n+                q.extend(cs)\n+                continue\n+            for obj in handler.tile(t.op.outputs):\n+                to_update_inputs = []\n+                chunks = []\n+                for inp in t.op.inputs:\n+                    chunks.extend(inp.chunks)\n+                    if has_unknown_shape(inp):\n+                        to_update_inputs.append(inp)\n+                if obj is None:\n+                    yield chunks + to_update_inputs\n+                else:\n+                    yield obj + to_update_inputs\n         q.pop()\n \n     if not return_list:\n", "issue": "[BUG] Duplicate operands execution\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\n\r\nThe case is `mars.core.entity.tests.test_utils.test_recursive_tile`. It generates a chunk graph with duplicate operands (the op key are same, the execution output are same).\r\n\r\nI added some code to check the operands in stage.py:\r\n\r\n``` python\r\nclass TaskStageProcessor:\r\n    def __init__(\r\n        self,\r\n        stage_id: str,\r\n        task: Task,\r\n        chunk_graph: ChunkGraph,\r\n        subtask_graph: SubtaskGraph,\r\n        bands: List[BandType],\r\n        tile_context: Dict[TileableType, TileableType],\r\n        scheduling_api: SchedulingAPI,\r\n        meta_api: MetaAPI,\r\n    ):\r\n        self.stage_id = stage_id\r\n        self.task = task\r\n        self.chunk_graph = chunk_graph\r\n        self.subtask_graph = subtask_graph\r\n        self._bands = bands\r\n        self._tile_context = tile_context\r\n\r\n        # Check for duplicate operand.\r\n        from collections import Counter\r\n        unique_keys = Counter(c.key for c in chunk_graph)\r\n        if len(unique_keys) < len(chunk_graph):\r\n            import pprint\r\n            pprint.pprint([c for c in chunk_graph])\r\n            pprint.pprint([k for k, v in unique_keys.items() if v > 1])\r\n            subtask_graph.view()\r\n```\r\n\r\nThe output is:\r\n\r\n```python\r\n[Chunk <op=TensorRand, key=560bdcf1ae4978a999cbe3fd620c65d7_0>,\r\n Chunk <op=TensorRand, key=5aed4a149f5084ce95b9212afb61f24d_0>,\r\n Chunk <op=TensorRand, key=6fa461793a5d0feef9905c945653653f_0>,\r\n Chunk <op=TensorRand, key=688bd5f2d8fe39ce3e1a8af91b53b65c_0>,\r\n Chunk <op=TensorSlice, key=a9cd6ffd308ed11c3fccd2753f9a4aba_0>,\r\n Chunk <op=PSRSShuffle, stage=reduce, key=94c1f8fa627bcd5960ad795f7c692671_0>,\r\n Chunk <op=TensorShuffleProxy, key=6e7794b968faabcbebdfb289789c1ab9_0>,\r\n Chunk <op=PSRSShuffle, stage=map, key=0ebf73717ef2479456a6351c6209a90e_0>,\r\n Chunk <op=PSRSShuffle, stage=map, key=eb053dc8499bc7cdc12d8382f199e096_0>,\r\n Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_0>,\r\n Chunk <op=PSRSConcatPivot, key=58ce89bb01132fd0fd757246ef654c40_0>,\r\n Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_1>,\r\n Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_1>,\r\n Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_0>,\r\n Chunk <op=TensorSlice, key=bdbeadf9202694d9c8f86b9fc8fc2db2_0>,\r\n Chunk <op=PSRSShuffle, stage=reduce, key=22974984dc0b50cff4d077e0413202d2_0>,\r\n Chunk <op=PSRSShuffle, stage=reduce, key=6c78a27cd13360463638ae139958297e_0>,\r\n Chunk <op=TensorShuffleProxy, key=fd341cca1ae125e359e9a2274928605f_0>,\r\n Chunk <op=PSRSShuffle, stage=map, key=7d9f9ea924cf86e2aba771257f46db95_0>,\r\n Chunk <op=PSRSShuffle, stage=map, key=45359c01dfab34dd6d552687e3e9cf86_0>,\r\n Chunk <op=PSRSSortRegularSample, key=cac04cc478ecb467c0f1a7bca4bf809c_0>,\r\n Chunk <op=PSRSConcatPivot, key=ee1f386a09653687d7f3802aaf6d8cd3_0>,\r\n Chunk <op=PSRSSortRegularSample, key=321f2eda3a690282eb2d75ea0aa2614e_1>,\r\n Chunk <op=PSRSSortRegularSample, key=cac04cc478ecb467c0f1a7bca4bf809c_1>,\r\n Chunk <op=PSRSSortRegularSample, key=321f2eda3a690282eb2d75ea0aa2614e_0>,\r\n Chunk <op=PSRSShuffle, stage=reduce, key=02d268f3091f3c95271acc314e0ef0fd_0>,\r\n Chunk <op=PSRSShuffle, stage=reduce, key=94c1f8fa627bcd5960ad795f7c692671_0>,\r\n Chunk <op=TensorShuffleProxy, key=6e7794b968faabcbebdfb289789c1ab9_0>,\r\n Chunk <op=PSRSShuffle, stage=map, key=0ebf73717ef2479456a6351c6209a90e_0>,\r\n Chunk <op=PSRSShuffle, stage=map, key=eb053dc8499bc7cdc12d8382f199e096_0>,\r\n Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_0>,\r\n Chunk <op=PSRSConcatPivot, key=58ce89bb01132fd0fd757246ef654c40_0>,\r\n Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_1>,\r\n Chunk <op=PSRSSortRegularSample, key=125e3a7f78bcb6c939668f03a3b94787_1>,\r\n Chunk <op=PSRSSortRegularSample, key=98e601719051e980b4d9f3f20b20e437_0>,\r\n Chunk <op=PSRSShuffle, stage=reduce, key=22974984dc0b50cff4d077e0413202d2_0>]\r\n['94c1f8fa627bcd5960ad795f7c692671_0',\r\n '6e7794b968faabcbebdfb289789c1ab9_0',\r\n '0ebf73717ef2479456a6351c6209a90e_0',\r\n 'eb053dc8499bc7cdc12d8382f199e096_0',\r\n '125e3a7f78bcb6c939668f03a3b94787_0',\r\n '58ce89bb01132fd0fd757246ef654c40_0',\r\n '98e601719051e980b4d9f3f20b20e437_1',\r\n '125e3a7f78bcb6c939668f03a3b94787_1',\r\n '98e601719051e980b4d9f3f20b20e437_0',\r\n '22974984dc0b50cff4d077e0413202d2_0']\r\n```\r\n\r\nDump graph: [default.pdf](https://github.com/mars-project/mars/files/8645133/default.pdf)\r\n\r\nI added check output code to the `PSRSShuffle._execute_map`:\r\n```python\r\nfor i in range(op.n_partition):\r\n    key = (out.key, (i,))\r\n    value = tuple(reduce_outputs[i].ravel())\r\n    if key in ctx:\r\n        existing_value = ctx[key]\r\n        np.testing.assert_array_equal(existing_value, value)\r\n        raise Exception(f\"The key already exists: {key}\")\r\n    ctx[out.key, (i,)] = tuple(reduce_outputs[i].ravel())\r\n```\r\n\r\nIt raises an exception and the `assert_array_equal` passes.\r\n\r\n```python\r\n            for i in range(op.n_partition):\r\n                key = (out.key, (i,))\r\n                value = tuple(reduce_outputs[i].ravel())\r\n                if key in ctx:\r\n                    existing_value = ctx[key]\r\n                    np.testing.assert_array_equal(existing_value, value)\r\n>                   raise Exception(f\"The key already exists: {key}\")\r\nE                   Exception: The key already exists: ('eb053dc8499bc7cdc12d8382f199e096_0', (0,))\r\n\r\nmars/tensor/base/psrs.py:753: Exception\r\n```\r\n\r\n**To Reproduce**\r\nTo help us reproducing this bug, please provide information below:\r\n1. Your Python version 3.7.7\r\n2. The version of Mars you use Latest master\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\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": "# 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\nfrom typing import List, Union, Generator\n\nfrom ...typing import TileableType, ChunkType\nfrom ...utils import has_unknown_shape, calc_nsplits\n\n\ndef refresh_tileable_shape(tileable):\n    if has_unknown_shape(tileable):\n        # update shape\n        nsplits = calc_nsplits({c.index: c.shape for c in tileable.chunks})\n        shape = tuple(sum(ns) for ns in nsplits)\n        tileable._nsplits = nsplits\n        tileable._shape = shape\n\n\ndef tile(tileable, *tileables: TileableType):\n    from ..graph import (\n        TileableGraph,\n        TileableGraphBuilder,\n        ChunkGraphBuilder,\n        TileContext,\n    )\n\n    raw_tileables = target_tileables = [tileable] + list(tileables)\n    target_tileables = [t.data if hasattr(t, \"data\") else t for t in target_tileables]\n\n    tileable_graph = TileableGraph(target_tileables)\n    tileable_graph_builder = TileableGraphBuilder(tileable_graph)\n    next(tileable_graph_builder.build())\n\n    # tile\n    tile_context = TileContext()\n    chunk_graph_builder = ChunkGraphBuilder(\n        tileable_graph, fuse_enabled=False, tile_context=tile_context\n    )\n    next(chunk_graph_builder.build())\n\n    if len(tileables) == 0:\n        return type(tileable)(tile_context[target_tileables[0]])\n    else:\n        return [\n            type(raw_t)(tile_context[t])\n            for raw_t, t in zip(raw_tileables, target_tileables)\n        ]\n\n\ndef recursive_tile(\n    tileable: TileableType, *tileables: TileableType\n) -> Generator[\n    List[ChunkType], List[ChunkType], Union[TileableType, List[TileableType]]\n]:\n    from .tileables import handler\n\n    return_list = len(tileables) > 0\n    if not return_list and isinstance(tileable, (list, tuple)):\n        return_list = True\n        raw = tileable\n        tileable = raw[0]\n        tileables = raw[1:]\n\n    to_tile = [tileable] + list(tileables)\n    q = [t for t in to_tile if t.is_coarse()]\n    while q:\n        t = q[-1]\n        cs = [c for c in t.inputs if c.is_coarse()]\n        if cs:\n            q.extend(cs)\n            continue\n        for obj in handler.tile(t.op.outputs):\n            to_update_inputs = []\n            chunks = []\n            for inp in t.op.inputs:\n                chunks.extend(inp.chunks)\n                if has_unknown_shape(inp):\n                    to_update_inputs.append(inp)\n            if obj is None:\n                yield chunks + to_update_inputs\n            else:\n                yield obj + to_update_inputs\n        q.pop()\n\n    if not return_list:\n        return tileable\n    else:\n        return [tileable] + list(tileables)\n", "path": "mars/core/entity/utils.py"}], "after_files": [{"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\nfrom typing import List, Union, Generator\n\nfrom ...typing import TileableType, ChunkType\nfrom ...utils import has_unknown_shape, calc_nsplits\n\n\ndef refresh_tileable_shape(tileable):\n    if has_unknown_shape(tileable):\n        # update shape\n        nsplits = calc_nsplits({c.index: c.shape for c in tileable.chunks})\n        shape = tuple(sum(ns) for ns in nsplits)\n        tileable._nsplits = nsplits\n        tileable._shape = shape\n\n\ndef tile(tileable, *tileables: TileableType):\n    from ..graph import (\n        TileableGraph,\n        TileableGraphBuilder,\n        ChunkGraphBuilder,\n        TileContext,\n    )\n\n    raw_tileables = target_tileables = [tileable] + list(tileables)\n    target_tileables = [t.data if hasattr(t, \"data\") else t for t in target_tileables]\n\n    tileable_graph = TileableGraph(target_tileables)\n    tileable_graph_builder = TileableGraphBuilder(tileable_graph)\n    next(tileable_graph_builder.build())\n\n    # tile\n    tile_context = TileContext()\n    chunk_graph_builder = ChunkGraphBuilder(\n        tileable_graph, fuse_enabled=False, tile_context=tile_context\n    )\n    next(chunk_graph_builder.build())\n\n    if len(tileables) == 0:\n        return type(tileable)(tile_context[target_tileables[0]])\n    else:\n        return [\n            type(raw_t)(tile_context[t])\n            for raw_t, t in zip(raw_tileables, target_tileables)\n        ]\n\n\ndef recursive_tile(\n    tileable: TileableType, *tileables: TileableType\n) -> Generator[\n    List[ChunkType], List[ChunkType], Union[TileableType, List[TileableType]]\n]:\n    from .tileables import handler\n\n    return_list = len(tileables) > 0\n    if not return_list and isinstance(tileable, (list, tuple)):\n        return_list = True\n        raw = tileable\n        tileable = raw[0]\n        tileables = raw[1:]\n\n    to_tile = [tileable] + list(tileables)\n    q = [t for t in to_tile if t.is_coarse()]\n    while q:\n        t = q[-1]\n        if t.is_coarse():\n            # t may be put into q repeatedly,\n            # so we check if it's tiled or not\n            cs = [c for c in t.inputs if c.is_coarse()]\n            if cs:\n                q.extend(cs)\n                continue\n            for obj in handler.tile(t.op.outputs):\n                to_update_inputs = []\n                chunks = []\n                for inp in t.op.inputs:\n                    chunks.extend(inp.chunks)\n                    if has_unknown_shape(inp):\n                        to_update_inputs.append(inp)\n                if obj is None:\n                    yield chunks + to_update_inputs\n                else:\n                    yield obj + to_update_inputs\n        q.pop()\n\n    if not return_list:\n        return tileable\n    else:\n        return [tileable] + list(tileables)\n", "path": "mars/core/entity/utils.py"}]}