Fix recursive_tile that it may cause duplicated tile for one tileable (#3021)

Author: Xuye (Chris) Qin

mars/core/entity/tests/test_utils.py

@@ -44,7 +44,7 @@ def test_recursive_tile(setup):
     d2 = mt.random.rand(10, chunk_size=5)
     op = _TestOperand()
     t = op.new_tensor([d1, d2], dtype=d1.dtype, shape=(20,), order=d1.order)
-    t.execute()
+    t.execute(extra_config={"check_duplicated_operand_keys": True})
 
 
 class _TestOperandWithDuplicatedSubmission(TensorOperand, TensorOperandMixin):

mars/core/entity/utils.py

@@ -76,21 +76,24 @@ def recursive_tile(
     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:

mars/services/task/supervisor/tests/task_preprocessor.py

@@ -51,6 +51,9 @@ def __init__(self, *args, **kwargs):
         for key in _check_args:
             check_options[key] = kwargs.get(key, True)
         self._check_options = check_options
+        self._check_duplicated_operand_keys = bool(
+            kwargs.get("check_duplicated_operand_keys")
+        )
 
     def _get_done(self):
         return super()._get_done()
@@ -145,6 +148,13 @@ def analyze(
         stage_id: str,
         op_to_bands: Dict[str, BandType] = None,
     ) -> SubtaskGraph:
+        # check if duplicated operand keys exist
+        if self._check_duplicated_operand_keys and len(
+            {c.key for c in chunk_graph}
+        ) < len(
+            chunk_graph
+        ):  # pragma: no cover
+            raise AssertionError("Duplicated operands exist")
         # record shapes generated in tile
         for n in chunk_graph:
             self._raw_chunk_shapes[n.key] = getattr(n, "shape", None)