prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  [BUG] Run the example code hangs in distributed mode
  <!--
  Thank you for your contribution!
  
  Please review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.
  -->
  
  **Describe the bug**
  Create a Mars cluster and run the code in readme:
  ``` Python
  import mars.tensor as mt
  N = 200_000_000
  a = mt.random.uniform(-1, 1, size=(N, 2))
  print(((mt.linalg.norm(a, axis=1) < 1)
          .sum() * 4 / N).execute())
  ```
  
  it hangs and error be found in server client:
  ```
  2020-11-09 21:30:22,053 mars.scheduler.operands.common 97 ERROR    Attempt 1: Unexpected error KeyError occurred in executing operand 230bef1901408a5f9134f34444918898 in 11.238.146.2:35131
  Traceback (most recent call last):
    File "/home/admin/work/public-mars-0.5.4.zip/mars/promise.py", line 378, in _wrapped
      return func(*args, **kwargs)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/utils.py", line 365, in _wrapped
      return func(*args, **kwargs)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/worker/execution.py", line 564, in execute_graph
      quota_request = self._prepare_quota_request(session_id, graph_key)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/worker/execution.py", line 249, in _prepare_quota_request
      memory_estimations = self._estimate_calc_memory(session_id, graph_key)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/worker/execution.py", line 213, in _estimate_calc_memory
      res = executor.execute_graph(graph_record.graph, graph_record.chunk_targets, mock=True)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 690, in execute_graph
      res = graph_execution.execute(retval)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 571, in execute
      future.result()
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 186, in result
      raise self._exc_info[1] from None
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 198, in submit
      return self._MockResult(fn(*args, **kwargs))
    File "/home/admin/work/public-mars-0.5.4.zip/mars/utils.py", line 439, in _inner
      return func(*args, **kwargs)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 443, in _execute_operand
      Executor.handle(first_op, results, self._mock)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 641, in handle
      return runner(results, op)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/tensor/fuse/ne.py", line 75, in estimate_size
      estimate_fuse_size(ctx, op)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/tensor/fuse/core.py", line 49, in estimate_fuse_size
      results = executor.execute_graph(dag, output_keys, mock=True, no_intermediate=True)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 690, in execute_graph
      res = graph_execution.execute(retval)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 571, in execute
      future.result()
    File "/opt/conda/lib/python3.7/concurrent/futures/_base.py", line 428, in result
      return self.__get_result()
    File "/opt/conda/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
      raise self._exception
    File "/opt/conda/lib/python3.7/concurrent/futures/thread.py", line 57, in run
      result = self.fn(*self.args, **self.kwargs)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/utils.py", line 439, in _inner
      return func(*args, **kwargs)
    File "/home/admin/work/public-mars-0.5.4.zip/mars/executor.py", line 486, in _execute_operand
      del results[dep_key]
  KeyError: '94e11781368129674925eb2d4ae093bf'
  ```
  
  
  --- 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/tensor/fuse/core.py`
  Content:
  ```
  1 #!/usr/bin/env python
  2 # -*- coding: utf-8 -*-
  3 # Copyright 1999-2020 Alibaba Group Holding Ltd.
  4 #
  5 # Licensed under the Apache License, Version 2.0 (the "License");
  6 # you may not use this file except in compliance with the License.
  7 # You may obtain a copy of the License at
  8 #
  9 #      http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 
  17 from ...operands import FuseChunkMixin
  18 from ..operands import TensorFuse, TensorOperandMixin
  19 
  20 
  21 class TensorFuseChunkMixin(FuseChunkMixin, TensorOperandMixin):
  22     __slots__ = ()
  23 
  24 
  25 class TensorFuseChunk(TensorFuse, TensorFuseChunkMixin):
  26     def __init__(self, dtype=None, **kw):
  27         super().__init__(_dtype=dtype, **kw)
  28 
  29 
  30 def estimate_fuse_size(ctx, op):
  31     from ...graph import DAG
  32     from ...executor import Executor
  33 
  34     chunk = op.outputs[0]
  35     dag = DAG()
  36     size_ctx = dict()
  37     keys = set(c.key for c in chunk.composed)
  38     for c in chunk.composed:
  39         dag.add_node(c)
  40         for inp in c.inputs:
  41             if inp.key not in keys:
  42                 size_ctx[inp.key] = ctx[inp.key]
  43             if inp not in dag:
  44                 dag.add_node(inp)
  45             dag.add_edge(inp, c)
  46 
  47     executor = Executor(storage=size_ctx)
  48     output_keys = [o.key for o in op.outputs]
  49     results = executor.execute_graph(dag, output_keys, mock=True, no_intermediate=True)
  50     ctx.update(zip(output_keys, results))
  51 
  52     # update with the maximal memory cost during the whole execution
  53     total_mem = sum(ctx[key][1] for key in output_keys)
  54     if total_mem:
  55         for key in output_keys:
  56             r = ctx[key]
  57             ctx[key] = (r[0], max(r[1], r[1] * executor.mock_max_memory // total_mem))
  58
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/mars/tensor/fuse/core.py b/mars/tensor/fuse/core.py
  --- a/mars/tensor/fuse/core.py
  +++ b/mars/tensor/fuse/core.py
  @@ -30,6 +30,7 @@
   def estimate_fuse_size(ctx, op):
       from ...graph import DAG
       from ...executor import Executor
  +    from ...utils import build_fetch_chunk
   
       chunk = op.outputs[0]
       dag = DAG()
  @@ -40,6 +41,7 @@
           for inp in c.inputs:
               if inp.key not in keys:
                   size_ctx[inp.key] = ctx[inp.key]
  +                inp = build_fetch_chunk(inp).data
               if inp not in dag:
                   dag.add_node(inp)
               dag.add_edge(inp, c)
  
verification_info: {"golden_diff": "diff --git a/mars/tensor/fuse/core.py b/mars/tensor/fuse/core.py\n--- a/mars/tensor/fuse/core.py\n+++ b/mars/tensor/fuse/core.py\n@@ -30,6 +30,7 @@\n def estimate_fuse_size(ctx, op):\n     from ...graph import DAG\n     from ...executor import Executor\n+    from ...utils import build_fetch_chunk\n \n     chunk = op.outputs[0]\n     dag = DAG()\n@@ -40,6 +41,7 @@\n         for inp in c.inputs:\n             if inp.key not in keys:\n                 size_ctx[inp.key] = ctx[inp.key]\n+                inp = build_fetch_chunk(inp).data\n             if inp not in dag:\n                 dag.add_node(inp)\n             dag.add_edge(inp, c)\n", "issue": "[BUG] Run the example code hangs in distributed mode\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\nCreate a Mars cluster and run the code in readme:\r\n``` Python\r\nimport mars.tensor as mt\r\nN = 200_000_000\r\na = mt.random.uniform(-1, 1, size=(N, 2))\r\nprint(((mt.linalg.norm(a, axis=1) < 1)\r\n        .sum() * 4 / N).execute())\r\n```\r\n\r\nit hangs and error be found in server client:\r\n```\r\n2020-11-09 21:30:22,053 mars.scheduler.operands.common 97 ERROR    Attempt 1: Unexpected error KeyError occurred in executing operand 230bef1901408a5f9134f34444918898 in 11.238.146.2:35131\r\nTraceback (most recent call last):\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/promise.py\", line 378, in _wrapped\r\n    return func(*args, **kwargs)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/utils.py\", line 365, in _wrapped\r\n    return func(*args, **kwargs)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/worker/execution.py\", line 564, in execute_graph\r\n    quota_request = self._prepare_quota_request(session_id, graph_key)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/worker/execution.py\", line 249, in _prepare_quota_request\r\n    memory_estimations = self._estimate_calc_memory(session_id, graph_key)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/worker/execution.py\", line 213, in _estimate_calc_memory\r\n    res = executor.execute_graph(graph_record.graph, graph_record.chunk_targets, mock=True)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 690, in execute_graph\r\n    res = graph_execution.execute(retval)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 571, in execute\r\n    future.result()\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 186, in result\r\n    raise self._exc_info[1] from None\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 198, in submit\r\n    return self._MockResult(fn(*args, **kwargs))\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/utils.py\", line 439, in _inner\r\n    return func(*args, **kwargs)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 443, in _execute_operand\r\n    Executor.handle(first_op, results, self._mock)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 641, in handle\r\n    return runner(results, op)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/tensor/fuse/ne.py\", line 75, in estimate_size\r\n    estimate_fuse_size(ctx, op)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/tensor/fuse/core.py\", line 49, in estimate_fuse_size\r\n    results = executor.execute_graph(dag, output_keys, mock=True, no_intermediate=True)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 690, in execute_graph\r\n    res = graph_execution.execute(retval)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 571, in execute\r\n    future.result()\r\n  File \"/opt/conda/lib/python3.7/concurrent/futures/_base.py\", line 428, in result\r\n    return self.__get_result()\r\n  File \"/opt/conda/lib/python3.7/concurrent/futures/_base.py\", line 384, in __get_result\r\n    raise self._exception\r\n  File \"/opt/conda/lib/python3.7/concurrent/futures/thread.py\", line 57, in run\r\n    result = self.fn(*self.args, **self.kwargs)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/utils.py\", line 439, in _inner\r\n    return func(*args, **kwargs)\r\n  File \"/home/admin/work/public-mars-0.5.4.zip/mars/executor.py\", line 486, in _execute_operand\r\n    del results[dep_key]\r\nKeyError: '94e11781368129674925eb2d4ae093bf'\r\n```\r\n\n", "before_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n# Copyright 1999-2020 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 ...operands import FuseChunkMixin\nfrom ..operands import TensorFuse, TensorOperandMixin\n\n\nclass TensorFuseChunkMixin(FuseChunkMixin, TensorOperandMixin):\n    __slots__ = ()\n\n\nclass TensorFuseChunk(TensorFuse, TensorFuseChunkMixin):\n    def __init__(self, dtype=None, **kw):\n        super().__init__(_dtype=dtype, **kw)\n\n\ndef estimate_fuse_size(ctx, op):\n    from ...graph import DAG\n    from ...executor import Executor\n\n    chunk = op.outputs[0]\n    dag = DAG()\n    size_ctx = dict()\n    keys = set(c.key for c in chunk.composed)\n    for c in chunk.composed:\n        dag.add_node(c)\n        for inp in c.inputs:\n            if inp.key not in keys:\n                size_ctx[inp.key] = ctx[inp.key]\n            if inp not in dag:\n                dag.add_node(inp)\n            dag.add_edge(inp, c)\n\n    executor = Executor(storage=size_ctx)\n    output_keys = [o.key for o in op.outputs]\n    results = executor.execute_graph(dag, output_keys, mock=True, no_intermediate=True)\n    ctx.update(zip(output_keys, results))\n\n    # update with the maximal memory cost during the whole execution\n    total_mem = sum(ctx[key][1] for key in output_keys)\n    if total_mem:\n        for key in output_keys:\n            r = ctx[key]\n            ctx[key] = (r[0], max(r[1], r[1] * executor.mock_max_memory // total_mem))\n", "path": "mars/tensor/fuse/core.py"}], "after_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n# Copyright 1999-2020 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 ...operands import FuseChunkMixin\nfrom ..operands import TensorFuse, TensorOperandMixin\n\n\nclass TensorFuseChunkMixin(FuseChunkMixin, TensorOperandMixin):\n    __slots__ = ()\n\n\nclass TensorFuseChunk(TensorFuse, TensorFuseChunkMixin):\n    def __init__(self, dtype=None, **kw):\n        super().__init__(_dtype=dtype, **kw)\n\n\ndef estimate_fuse_size(ctx, op):\n    from ...graph import DAG\n    from ...executor import Executor\n    from ...utils import build_fetch_chunk\n\n    chunk = op.outputs[0]\n    dag = DAG()\n    size_ctx = dict()\n    keys = set(c.key for c in chunk.composed)\n    for c in chunk.composed:\n        dag.add_node(c)\n        for inp in c.inputs:\n            if inp.key not in keys:\n                size_ctx[inp.key] = ctx[inp.key]\n                inp = build_fetch_chunk(inp).data\n            if inp not in dag:\n                dag.add_node(inp)\n            dag.add_edge(inp, c)\n\n    executor = Executor(storage=size_ctx)\n    output_keys = [o.key for o in op.outputs]\n    results = executor.execute_graph(dag, output_keys, mock=True, no_intermediate=True)\n    ctx.update(zip(output_keys, results))\n\n    # update with the maximal memory cost during the whole execution\n    total_mem = sum(ctx[key][1] for key in output_keys)\n    if total_mem:\n        for key in output_keys:\n            r = ctx[key]\n            ctx[key] = (r[0], max(r[1], r[1] * executor.mock_max_memory // total_mem))\n", "path": "mars/tensor/fuse/core.py"}]}