Fix executor's behavior of deletes & bump version (#249)

Author: hekaisheng

mars/_version.py

@@ -16,7 +16,7 @@
 import os
 import sys
 
-version_info = (0, 2, 0, 'a1')
+version_info = (0, 2, 0, 'a2')
 _num_index = max(idx if isinstance(v, int) else 0
                  for idx, v in enumerate(version_info))
 __version__ = '.'.join(map(str, version_info[:_num_index + 1])) + \

mars/executor.py

@@ -219,6 +219,8 @@ def _execute_operand(self, op):
         results = self._chunk_results
         ref_counts = self._chunk_key_ref_counts
         op_keys = self._executed_op_keys
+        executed_chunk_keys = set()
+        deleted_chunk_keys = set()
         try:
             ops = list(self._op_key_to_ops[op.key])
             if not self._mock:
@@ -227,11 +229,15 @@ def _execute_operand(self, op):
                 # so we pass the first operand's first output to Executor.handle
                 first_op = ops[0]
                 Executor.handle(first_op.outputs[0], results)
+                executed_chunk_keys.update([c.key for c in first_op.outputs])
                 op_keys.add(first_op.key)
                 # handle other operands
                 for rest_op in ops[1:]:
                     for op_output, rest_op_output in zip(first_op.outputs, rest_op.outputs):
-                        results[rest_op_output.key] = results[op_output.key]
+                        # if the op's outputs have been stored,
+                        # other same key ops' results will be the same
+                        if rest_op_output.key not in executed_chunk_keys:
+                            results[rest_op_output.key] = results[op_output.key]
             else:
                 sparse_percent = self._sparse_mock_percent if op.sparse else 1.0
                 for output in op.outputs:
@@ -245,7 +251,10 @@ def _execute_operand(self, op):
                     # in case that operand has multiple outputs
                     # and some of the output not in result keys, delete them
                     if ref_counts.get(output.key) == 0:
-                        del results[output.key]
+                        # if the result has been deleted, it should be skipped
+                        if output.key not in deleted_chunk_keys:
+                            deleted_chunk_keys.add(output.key)
+                            del results[output.key]
 
                     # clean the predecessors' results if ref counts equals 0
                     for pred_chunk in self._graph.iter_predecessors(output):

mars/tensor/execution/tests/test_linalg_execute.py

@@ -112,6 +112,15 @@ def testSVDExecution(self):
         res = self.executor.execute_tensor(t, concat=True)[0]
         self.assertTrue(np.allclose(res, data))
 
+        # test for matrix of ones
+        data = np.ones((20, 10))
+
+        a = tensor(data, chunk_size=10)
+        s = svd(a)[1]
+        res = self.executor.execute_tensor(s, concat=True)[0]
+        expected = np.linalg.svd(a)[1]
+        np.testing.assert_array_almost_equal(res, expected)
+
     def testCholeskyExecution(self):
         data = np.array([[1, -2j], [2j, 5]])