We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG][mars on ray] Failed to run in ray cluster mode.
**Describe the bug**
Ray cluster will load code from local (not deserialize from bytes), but mars `RemoteMetaStore` is nested in `RayStorage` class. Ray worker can't load the `RemoteMetaStore` from local code.

```
2020-11-17 16:48:29,349	WARNING worker.py:1157 -- Traceback (most recent call last):
  File "/home/admin/.local/lib/python3.6/site-packages/ray/function_manager.py", line 445, in _load_actor_class_from_local
    actor_class = getattr(module, class_name)
AttributeError: module 'mars.ray.core' has no attribute 'RemoteMetaStore'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "python/ray/_raylet.pyx", line 563, in ray._raylet.task_execution_handler
  File "python/ray/_raylet.pyx", line 567, in ray._raylet.task_execution_handler
  File "python/ray/_raylet.pyx", line 364, in ray._raylet.execute_task
  File "/home/admin/.local/lib/python3.6/site-packages/ray/function_manager.py", line 394, in load_actor_class
    job_id, actor_creation_function_descriptor)
  File "/home/admin/.local/lib/python3.6/site-packages/ray/function_manager.py", line 454, in _load_actor_class_from_local
    class_name))
RuntimeError: Actor RemoteMetaStore failed to be imported from local code.
An unexpected internal error occurred while the worker was executing a task.
```

**To Reproduce**
To help us reproducing this bug, please provide information below:
1. Python 3.6.5
2. pymars 0.6.0b2
3. ray 1.0.1
4. Minimal reproduce example.
``` python
from mars.session import new_session

ray_session = new_session(backend='ray', _load_code_from_local=True).as_default()
print(ray_session)


def main():
    import mars.dataframe as md
    import mars.tensor as mt
    t = mt.random.rand(100, 4, chunk_size=30)
    df = md.DataFrame(t, columns=list('abcd'))
    print(df.describe().execute())


if __name__ == "__main__":
    main()
```

**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/ray/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 import uuid
18 from collections import namedtuple
19 from functools import lru_cache
20 from typing import Dict
21 
22 import ray
23 
24 from ..graph import DAG
25 from ..operands import Fetch
26 from ..tiles import get_tiled
27 from ..utils import build_fetch_chunk
28 from ..executor import Executor, GraphExecution
29 
30 
31 class _OperandWrapper:
32     __slots__ = 'op', 'chunks'
33 
34     def __init__(self, op, chunks):
35         """
36         As we only serde op for Ray executors, but op only weakly reference chunks,
37         So we create a wrapper here to keep the reference
38         """
39         self.op = op
40         self.chunks = chunks
41 
42 
43 def operand_serializer(op):
44     graph = DAG()
45     inputs = [build_fetch_chunk(inp) for inp in op.inputs or []]
46     new_op = op.copy()
47 
48     kws = []
49     for c in op.outputs:
50         params = c.params.copy()
51         params['_key'] = c.key
52         params.update(c.extra_params)
53         kws.append(params)
54 
55     chunks = new_op.new_chunks(inputs, kws=kws, output_limit=len(kws))
56     for obj in chunks + inputs:
57         graph.add_node(obj)
58 
59     return graph.to_json()
60 
61 
62 def operand_deserializer(value):
63     graph = DAG.from_json(value)
64     if len(graph) == 1:
65         chunks = [list(graph)[0]]
66     else:
67         chunks = [c for c in graph if not isinstance(c.op, Fetch)]
68     op = chunks[0].op
69     return _OperandWrapper(op, chunks)
70 
71 
72 @lru_cache(500)
73 def _register_ray_serializer(op):
74     # register a custom serializer for Mars operand
75     try:
76         ray.register_custom_serializer(
77             type(op), serializer=operand_serializer,
78             deserializer=operand_deserializer)
79     except AttributeError:  # ray >= 1.0
80         from ray.worker import global_worker
81 
82         global_worker.check_connected()
83         context = global_worker.get_serialization_context()
84         context.register_custom_serializer(
85             type(op), serializer=operand_serializer,
86             deserializer=operand_deserializer)
87 
88 
89 class GraphExecutionForRay(GraphExecution):
90     def handle_op(self, *args, **kw):
91         return RayExecutor.handle(*args, **kw)
92 
93 
94 ChunkMeta = namedtuple('ChunkMeta', ['shape', 'object_id'])
95 
96 
97 class RayStorage:
98     """
99     `RayStorage` is a dict-like class. When executed in local, Mars executor will store chunk result in a
100     dict(chunk_key -> chunk_result), here uses Ray actor to store them as remote objects.
101     """
102 
103     @ray.remote
104     class RemoteMetaStore:
105         def __init__(self):
106             self._store = dict()
107 
108         def set_meta(self, chunk_key, meta):
109             self._store[chunk_key] = meta
110 
111         def get_meta(self, chunk_key):
112             return self._store[chunk_key]
113 
114         def get_shape(self, chunk_key):
115             return self._store[chunk_key].shape
116 
117         def chunk_keys(self):
118             return list(self._store.keys())
119 
120         def delete_keys(self, keys):
121             if not isinstance(keys, (list, tuple)):
122                 keys = [keys]
123             for k in keys:
124                 del self._store[k]
125 
126     def __init__(self, meta_store=None):
127         self.meta_store = meta_store or RayStorage.RemoteMetaStore.remote()
128 
129     def __getitem__(self, item):
130         meta: ChunkMeta = ray.get(self.meta_store.get_meta.remote(item))
131         return ray.get(meta.object_id)
132 
133     def __setitem__(self, key, value):
134         object_id = ray.put(value)
135         shape = getattr(value, 'shape', None)
136         meta = ChunkMeta(shape=shape, object_id=object_id)
137         set_meta = self.meta_store.set_meta.remote(key, meta)
138         ray.wait([object_id, set_meta])
139 
140     def copy(self):
141         return RayStorage(meta_store=self.meta_store)
142 
143     def update(self, mapping: Dict):
144         tasks = []
145         for k, v in mapping.items():
146             object_id = ray.put(v)
147             tasks.append(object_id)
148             shape = getattr(v, 'shape', None)
149             meta = ChunkMeta(shape=shape, object_id=object_id)
150             set_meta = self.meta_store.set_meta.remote(k, meta)
151             tasks.append(set_meta)
152         ray.wait(tasks)
153 
154     def __iter__(self):
155         return iter(ray.get(self.meta_store.chunk_keys.remote()))
156 
157     def __delitem__(self, key):
158         ray.wait([self.meta_store.delete_keys.remote(key)])
159 
160 
161 @ray.remote
162 def execute_on_ray(func, results, op_wrapper: _OperandWrapper):
163     op = op_wrapper.op
164     func(results, op)
165 
166 
167 class RayExecutor(Executor):
168     """
169     Wraps the execute function as a Ray remote function, the type of `results` is `RayStorage`,
170     when operand is executed, it will fetch dependencies from a Ray actor.
171     """
172 
173     _graph_execution_cls = GraphExecutionForRay
174 
175     @classmethod
176     def handle(cls, op, results, mock=False):
177         method_name, mapper = ('execute', cls._op_runners) if not mock else \
178             ('estimate_size', cls._op_size_estimators)
179         try:
180             runner = mapper[type(op)]
181         except KeyError:
182             runner = getattr(op, method_name)
183 
184         # register a custom serializer for Mars operand
185         _register_ray_serializer(op)
186 
187         try:
188             ray.wait([execute_on_ray.remote(runner, results, op)])
189         except NotImplementedError:
190             for op_cls in mapper.keys():
191                 if isinstance(op, op_cls):
192                     mapper[type(op)] = mapper[op_cls]
193                     runner = mapper[op_cls]
194 
195                     ray.wait(
196                         [execute_on_ray.remote(runner, results, op)])
197             raise KeyError(f'No handler found for op: {op}')
198 
199     @classmethod
200     def _get_chunk_shape(cls, chunk_key, chunk_result):
201         assert isinstance(chunk_result, RayStorage)
202         return ray.get(chunk_result.meta_store.get_shape.remote(chunk_key))
203 
204 
205 class RaySession:
206     """
207     Session to submit Mars job to Ray cluster.
208 
209     If Ray is not initialized, kwargs will pass to initialize Ray.
210     """
211     def __init__(self, **kwargs):
212         # as we cannot serialize fuse chunk for now,
213         # we just disable numexpr for ray executor
214         engine = kwargs.pop('engine', ['numpy', 'dataframe'])
215         if not ray.is_initialized():
216             ray.init(**kwargs)
217         self._session_id = uuid.uuid4()
218         self._executor = RayExecutor(engine=engine,
219                                      storage=RayStorage())
220 
221     @property
222     def session_id(self):
223         return self._session_id
224 
225     @property
226     def executor(self):
227         return self._executor
228 
229     def fetch(self, *tileables, **kw):
230         return self._executor.fetch_tileables(tileables, **kw)
231 
232     def fetch_log(self, tileables, offsets=None, sizes=None):  # pragma: no cover
233         raise NotImplementedError('`fetch_log` is not implemented for ray executor')
234 
235     def run(self, *tileables, **kw):
236         """
237         Parallelism equals to Ray cluster CPUs.
238         """
239         if 'n_parallel' not in kw:  # pragma: no cover
240             kw['n_parallel'] = ray.cluster_resources()['CPU']
241         return self._executor.execute_tileables(tileables, **kw)
242 
243     def _update_tileable_shape(self, tileable):
244         from ..optimizes.tileable_graph import tileable_optimized
245 
246         new_nsplits = self._executor.get_tileable_nsplits(tileable)
247         tiled = get_tiled(tileable, mapping=tileable_optimized)
248         for t in (tileable, tiled):
249             t._update_shape(tuple(sum(nsplit) for nsplit in new_nsplits))
250         tiled.nsplits = new_nsplits
251 
252     def __enter__(self):
253         return self
254 
255     def __exit__(self, *_):
256         self._executor = None
257
```
--- 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/ray/core.py b/mars/ray/core.py
--- a/mars/ray/core.py
+++ b/mars/ray/core.py
@@ -94,37 +94,37 @@
 ChunkMeta = namedtuple('ChunkMeta', ['shape', 'object_id'])
 
 
-class RayStorage:
-    """
-    `RayStorage` is a dict-like class. When executed in local, Mars executor will store chunk result in a
-    dict(chunk_key -> chunk_result), here uses Ray actor to store them as remote objects.
-    """
+@ray.remote
+class RemoteMetaStore:
+    def __init__(self):
+        self._store = dict()
 
-    @ray.remote
-    class RemoteMetaStore:
-        def __init__(self):
-            self._store = dict()
+    def set_meta(self, chunk_key, meta):
+        self._store[chunk_key] = meta
 
-        def set_meta(self, chunk_key, meta):
-            self._store[chunk_key] = meta
+    def get_meta(self, chunk_key):
+        return self._store[chunk_key]
 
-        def get_meta(self, chunk_key):
-            return self._store[chunk_key]
+    def get_shape(self, chunk_key):
+        return self._store[chunk_key].shape
 
-        def get_shape(self, chunk_key):
-            return self._store[chunk_key].shape
+    def chunk_keys(self):
+        return list(self._store.keys())
 
-        def chunk_keys(self):
-            return list(self._store.keys())
+    def delete_keys(self, keys):
+        if not isinstance(keys, (list, tuple)):
+            keys = [keys]
+        for k in keys:
+            del self._store[k]
 
-        def delete_keys(self, keys):
-            if not isinstance(keys, (list, tuple)):
-                keys = [keys]
-            for k in keys:
-                del self._store[k]
 
+class RayStorage:
+    """
+    `RayStorage` is a dict-like class. When executed in local, Mars executor will store chunk result in a
+    dict(chunk_key -> chunk_result), here uses Ray actor to store them as remote objects.
+    """
     def __init__(self, meta_store=None):
-        self.meta_store = meta_store or RayStorage.RemoteMetaStore.remote()
+        self.meta_store = meta_store or RemoteMetaStore.remote()
 
     def __getitem__(self, item):
         meta: ChunkMeta = ray.get(self.meta_store.get_meta.remote(item))

{"golden_diff": "diff --git a/mars/ray/core.py b/mars/ray/core.py\n--- a/mars/ray/core.py\n+++ b/mars/ray/core.py\n@@ -94,37 +94,37 @@\n ChunkMeta = namedtuple('ChunkMeta', ['shape', 'object_id'])\n \n \n-class RayStorage:\n-    \"\"\"\n-    `RayStorage` is a dict-like class. When executed in local, Mars executor will store chunk result in a\n-    dict(chunk_key -> chunk_result), here uses Ray actor to store them as remote objects.\n-    \"\"\"\n+@ray.remote\n+class RemoteMetaStore:\n+    def __init__(self):\n+        self._store = dict()\n \n-    @ray.remote\n-    class RemoteMetaStore:\n-        def __init__(self):\n-            self._store = dict()\n+    def set_meta(self, chunk_key, meta):\n+        self._store[chunk_key] = meta\n \n-        def set_meta(self, chunk_key, meta):\n-            self._store[chunk_key] = meta\n+    def get_meta(self, chunk_key):\n+        return self._store[chunk_key]\n \n-        def get_meta(self, chunk_key):\n-            return self._store[chunk_key]\n+    def get_shape(self, chunk_key):\n+        return self._store[chunk_key].shape\n \n-        def get_shape(self, chunk_key):\n-            return self._store[chunk_key].shape\n+    def chunk_keys(self):\n+        return list(self._store.keys())\n \n-        def chunk_keys(self):\n-            return list(self._store.keys())\n+    def delete_keys(self, keys):\n+        if not isinstance(keys, (list, tuple)):\n+            keys = [keys]\n+        for k in keys:\n+            del self._store[k]\n \n-        def delete_keys(self, keys):\n-            if not isinstance(keys, (list, tuple)):\n-                keys = [keys]\n-            for k in keys:\n-                del self._store[k]\n \n+class RayStorage:\n+    \"\"\"\n+    `RayStorage` is a dict-like class. When executed in local, Mars executor will store chunk result in a\n+    dict(chunk_key -> chunk_result), here uses Ray actor to store them as remote objects.\n+    \"\"\"\n     def __init__(self, meta_store=None):\n-        self.meta_store = meta_store or RayStorage.RemoteMetaStore.remote()\n+        self.meta_store = meta_store or RemoteMetaStore.remote()\n \n     def __getitem__(self, item):\n         meta: ChunkMeta = ray.get(self.meta_store.get_meta.remote(item))\n", "issue": "[BUG][mars on ray] Failed to run in ray cluster mode.\n**Describe the bug**\r\nRay cluster will load code from local (not deserialize from bytes), but mars `RemoteMetaStore` is nested in `RayStorage` class. Ray worker can't load the `RemoteMetaStore` from local code.\r\n\r\n```\r\n2020-11-17 16:48:29,349\tWARNING worker.py:1157 -- Traceback (most recent call last):\r\n  File \"/home/admin/.local/lib/python3.6/site-packages/ray/function_manager.py\", line 445, in _load_actor_class_from_local\r\n    actor_class = getattr(module, class_name)\r\nAttributeError: module 'mars.ray.core' has no attribute 'RemoteMetaStore'\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nTraceback (most recent call last):\r\n  File \"python/ray/_raylet.pyx\", line 563, in ray._raylet.task_execution_handler\r\n  File \"python/ray/_raylet.pyx\", line 567, in ray._raylet.task_execution_handler\r\n  File \"python/ray/_raylet.pyx\", line 364, in ray._raylet.execute_task\r\n  File \"/home/admin/.local/lib/python3.6/site-packages/ray/function_manager.py\", line 394, in load_actor_class\r\n    job_id, actor_creation_function_descriptor)\r\n  File \"/home/admin/.local/lib/python3.6/site-packages/ray/function_manager.py\", line 454, in _load_actor_class_from_local\r\n    class_name))\r\nRuntimeError: Actor RemoteMetaStore failed to be imported from local code.\r\nAn unexpected internal error occurred while the worker was executing a task.\r\n```\r\n\r\n**To Reproduce**\r\nTo help us reproducing this bug, please provide information below:\r\n1. Python 3.6.5\r\n2. pymars 0.6.0b2\r\n3. ray 1.0.1\r\n4. Minimal reproduce example.\r\n``` python\r\nfrom mars.session import new_session\r\n\r\nray_session = new_session(backend='ray', _load_code_from_local=True).as_default()\r\nprint(ray_session)\r\n\r\n\r\ndef main():\r\n    import mars.dataframe as md\r\n    import mars.tensor as mt\r\n    t = mt.random.rand(100, 4, chunk_size=30)\r\n    df = md.DataFrame(t, columns=list('abcd'))\r\n    print(df.describe().execute())\r\n\r\n\r\nif __name__ == \"__main__\":\r\n    main()\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\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\nimport uuid\nfrom collections import namedtuple\nfrom functools import lru_cache\nfrom typing import Dict\n\nimport ray\n\nfrom ..graph import DAG\nfrom ..operands import Fetch\nfrom ..tiles import get_tiled\nfrom ..utils import build_fetch_chunk\nfrom ..executor import Executor, GraphExecution\n\n\nclass _OperandWrapper:\n    __slots__ = 'op', 'chunks'\n\n    def __init__(self, op, chunks):\n        \"\"\"\n        As we only serde op for Ray executors, but op only weakly reference chunks,\n        So we create a wrapper here to keep the reference\n        \"\"\"\n        self.op = op\n        self.chunks = chunks\n\n\ndef operand_serializer(op):\n    graph = DAG()\n    inputs = [build_fetch_chunk(inp) for inp in op.inputs or []]\n    new_op = op.copy()\n\n    kws = []\n    for c in op.outputs:\n        params = c.params.copy()\n        params['_key'] = c.key\n        params.update(c.extra_params)\n        kws.append(params)\n\n    chunks = new_op.new_chunks(inputs, kws=kws, output_limit=len(kws))\n    for obj in chunks + inputs:\n        graph.add_node(obj)\n\n    return graph.to_json()\n\n\ndef operand_deserializer(value):\n    graph = DAG.from_json(value)\n    if len(graph) == 1:\n        chunks = [list(graph)[0]]\n    else:\n        chunks = [c for c in graph if not isinstance(c.op, Fetch)]\n    op = chunks[0].op\n    return _OperandWrapper(op, chunks)\n\n\n@lru_cache(500)\ndef _register_ray_serializer(op):\n    # register a custom serializer for Mars operand\n    try:\n        ray.register_custom_serializer(\n            type(op), serializer=operand_serializer,\n            deserializer=operand_deserializer)\n    except AttributeError:  # ray >= 1.0\n        from ray.worker import global_worker\n\n        global_worker.check_connected()\n        context = global_worker.get_serialization_context()\n        context.register_custom_serializer(\n            type(op), serializer=operand_serializer,\n            deserializer=operand_deserializer)\n\n\nclass GraphExecutionForRay(GraphExecution):\n    def handle_op(self, *args, **kw):\n        return RayExecutor.handle(*args, **kw)\n\n\nChunkMeta = namedtuple('ChunkMeta', ['shape', 'object_id'])\n\n\nclass RayStorage:\n    \"\"\"\n    `RayStorage` is a dict-like class. When executed in local, Mars executor will store chunk result in a\n    dict(chunk_key -> chunk_result), here uses Ray actor to store them as remote objects.\n    \"\"\"\n\n    @ray.remote\n    class RemoteMetaStore:\n        def __init__(self):\n            self._store = dict()\n\n        def set_meta(self, chunk_key, meta):\n            self._store[chunk_key] = meta\n\n        def get_meta(self, chunk_key):\n            return self._store[chunk_key]\n\n        def get_shape(self, chunk_key):\n            return self._store[chunk_key].shape\n\n        def chunk_keys(self):\n            return list(self._store.keys())\n\n        def delete_keys(self, keys):\n            if not isinstance(keys, (list, tuple)):\n                keys = [keys]\n            for k in keys:\n                del self._store[k]\n\n    def __init__(self, meta_store=None):\n        self.meta_store = meta_store or RayStorage.RemoteMetaStore.remote()\n\n    def __getitem__(self, item):\n        meta: ChunkMeta = ray.get(self.meta_store.get_meta.remote(item))\n        return ray.get(meta.object_id)\n\n    def __setitem__(self, key, value):\n        object_id = ray.put(value)\n        shape = getattr(value, 'shape', None)\n        meta = ChunkMeta(shape=shape, object_id=object_id)\n        set_meta = self.meta_store.set_meta.remote(key, meta)\n        ray.wait([object_id, set_meta])\n\n    def copy(self):\n        return RayStorage(meta_store=self.meta_store)\n\n    def update(self, mapping: Dict):\n        tasks = []\n        for k, v in mapping.items():\n            object_id = ray.put(v)\n            tasks.append(object_id)\n            shape = getattr(v, 'shape', None)\n            meta = ChunkMeta(shape=shape, object_id=object_id)\n            set_meta = self.meta_store.set_meta.remote(k, meta)\n            tasks.append(set_meta)\n        ray.wait(tasks)\n\n    def __iter__(self):\n        return iter(ray.get(self.meta_store.chunk_keys.remote()))\n\n    def __delitem__(self, key):\n        ray.wait([self.meta_store.delete_keys.remote(key)])\n\n\n@ray.remote\ndef execute_on_ray(func, results, op_wrapper: _OperandWrapper):\n    op = op_wrapper.op\n    func(results, op)\n\n\nclass RayExecutor(Executor):\n    \"\"\"\n    Wraps the execute function as a Ray remote function, the type of `results` is `RayStorage`,\n    when operand is executed, it will fetch dependencies from a Ray actor.\n    \"\"\"\n\n    _graph_execution_cls = GraphExecutionForRay\n\n    @classmethod\n    def handle(cls, op, results, mock=False):\n        method_name, mapper = ('execute', cls._op_runners) if not mock else \\\n            ('estimate_size', cls._op_size_estimators)\n        try:\n            runner = mapper[type(op)]\n        except KeyError:\n            runner = getattr(op, method_name)\n\n        # register a custom serializer for Mars operand\n        _register_ray_serializer(op)\n\n        try:\n            ray.wait([execute_on_ray.remote(runner, results, op)])\n        except NotImplementedError:\n            for op_cls in mapper.keys():\n                if isinstance(op, op_cls):\n                    mapper[type(op)] = mapper[op_cls]\n                    runner = mapper[op_cls]\n\n                    ray.wait(\n                        [execute_on_ray.remote(runner, results, op)])\n            raise KeyError(f'No handler found for op: {op}')\n\n    @classmethod\n    def _get_chunk_shape(cls, chunk_key, chunk_result):\n        assert isinstance(chunk_result, RayStorage)\n        return ray.get(chunk_result.meta_store.get_shape.remote(chunk_key))\n\n\nclass RaySession:\n    \"\"\"\n    Session to submit Mars job to Ray cluster.\n\n    If Ray is not initialized, kwargs will pass to initialize Ray.\n    \"\"\"\n    def __init__(self, **kwargs):\n        # as we cannot serialize fuse chunk for now,\n        # we just disable numexpr for ray executor\n        engine = kwargs.pop('engine', ['numpy', 'dataframe'])\n        if not ray.is_initialized():\n            ray.init(**kwargs)\n        self._session_id = uuid.uuid4()\n        self._executor = RayExecutor(engine=engine,\n                                     storage=RayStorage())\n\n    @property\n    def session_id(self):\n        return self._session_id\n\n    @property\n    def executor(self):\n        return self._executor\n\n    def fetch(self, *tileables, **kw):\n        return self._executor.fetch_tileables(tileables, **kw)\n\n    def fetch_log(self, tileables, offsets=None, sizes=None):  # pragma: no cover\n        raise NotImplementedError('`fetch_log` is not implemented for ray executor')\n\n    def run(self, *tileables, **kw):\n        \"\"\"\n        Parallelism equals to Ray cluster CPUs.\n        \"\"\"\n        if 'n_parallel' not in kw:  # pragma: no cover\n            kw['n_parallel'] = ray.cluster_resources()['CPU']\n        return self._executor.execute_tileables(tileables, **kw)\n\n    def _update_tileable_shape(self, tileable):\n        from ..optimizes.tileable_graph import tileable_optimized\n\n        new_nsplits = self._executor.get_tileable_nsplits(tileable)\n        tiled = get_tiled(tileable, mapping=tileable_optimized)\n        for t in (tileable, tiled):\n            t._update_shape(tuple(sum(nsplit) for nsplit in new_nsplits))\n        tiled.nsplits = new_nsplits\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, *_):\n        self._executor = None\n", "path": "mars/ray/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\nimport uuid\nfrom collections import namedtuple\nfrom functools import lru_cache\nfrom typing import Dict\n\nimport ray\n\nfrom ..graph import DAG\nfrom ..operands import Fetch\nfrom ..tiles import get_tiled\nfrom ..utils import build_fetch_chunk\nfrom ..executor import Executor, GraphExecution\n\n\nclass _OperandWrapper:\n    __slots__ = 'op', 'chunks'\n\n    def __init__(self, op, chunks):\n        \"\"\"\n        As we only serde op for Ray executors, but op only weakly reference chunks,\n        So we create a wrapper here to keep the reference\n        \"\"\"\n        self.op = op\n        self.chunks = chunks\n\n\ndef operand_serializer(op):\n    graph = DAG()\n    inputs = [build_fetch_chunk(inp) for inp in op.inputs or []]\n    new_op = op.copy()\n\n    kws = []\n    for c in op.outputs:\n        params = c.params.copy()\n        params['_key'] = c.key\n        params.update(c.extra_params)\n        kws.append(params)\n\n    chunks = new_op.new_chunks(inputs, kws=kws, output_limit=len(kws))\n    for obj in chunks + inputs:\n        graph.add_node(obj)\n\n    return graph.to_json()\n\n\ndef operand_deserializer(value):\n    graph = DAG.from_json(value)\n    if len(graph) == 1:\n        chunks = [list(graph)[0]]\n    else:\n        chunks = [c for c in graph if not isinstance(c.op, Fetch)]\n    op = chunks[0].op\n    return _OperandWrapper(op, chunks)\n\n\n@lru_cache(500)\ndef _register_ray_serializer(op):\n    # register a custom serializer for Mars operand\n    try:\n        ray.register_custom_serializer(\n            type(op), serializer=operand_serializer,\n            deserializer=operand_deserializer)\n    except AttributeError:  # ray >= 1.0\n        from ray.worker import global_worker\n\n        global_worker.check_connected()\n        context = global_worker.get_serialization_context()\n        context.register_custom_serializer(\n            type(op), serializer=operand_serializer,\n            deserializer=operand_deserializer)\n\n\nclass GraphExecutionForRay(GraphExecution):\n    def handle_op(self, *args, **kw):\n        return RayExecutor.handle(*args, **kw)\n\n\nChunkMeta = namedtuple('ChunkMeta', ['shape', 'object_id'])\n\n\n@ray.remote\nclass RemoteMetaStore:\n    def __init__(self):\n        self._store = dict()\n\n    def set_meta(self, chunk_key, meta):\n        self._store[chunk_key] = meta\n\n    def get_meta(self, chunk_key):\n        return self._store[chunk_key]\n\n    def get_shape(self, chunk_key):\n        return self._store[chunk_key].shape\n\n    def chunk_keys(self):\n        return list(self._store.keys())\n\n    def delete_keys(self, keys):\n        if not isinstance(keys, (list, tuple)):\n            keys = [keys]\n        for k in keys:\n            del self._store[k]\n\n\nclass RayStorage:\n    \"\"\"\n    `RayStorage` is a dict-like class. When executed in local, Mars executor will store chunk result in a\n    dict(chunk_key -> chunk_result), here uses Ray actor to store them as remote objects.\n    \"\"\"\n    def __init__(self, meta_store=None):\n        self.meta_store = meta_store or RemoteMetaStore.remote()\n\n    def __getitem__(self, item):\n        meta: ChunkMeta = ray.get(self.meta_store.get_meta.remote(item))\n        return ray.get(meta.object_id)\n\n    def __setitem__(self, key, value):\n        object_id = ray.put(value)\n        shape = getattr(value, 'shape', None)\n        meta = ChunkMeta(shape=shape, object_id=object_id)\n        set_meta = self.meta_store.set_meta.remote(key, meta)\n        ray.wait([object_id, set_meta])\n\n    def copy(self):\n        return RayStorage(meta_store=self.meta_store)\n\n    def update(self, mapping: Dict):\n        tasks = []\n        for k, v in mapping.items():\n            object_id = ray.put(v)\n            tasks.append(object_id)\n            shape = getattr(v, 'shape', None)\n            meta = ChunkMeta(shape=shape, object_id=object_id)\n            set_meta = self.meta_store.set_meta.remote(k, meta)\n            tasks.append(set_meta)\n        ray.wait(tasks)\n\n    def __iter__(self):\n        return iter(ray.get(self.meta_store.chunk_keys.remote()))\n\n    def __delitem__(self, key):\n        ray.wait([self.meta_store.delete_keys.remote(key)])\n\n\n@ray.remote\ndef execute_on_ray(func, results, op_wrapper: _OperandWrapper):\n    op = op_wrapper.op\n    func(results, op)\n\n\nclass RayExecutor(Executor):\n    \"\"\"\n    Wraps the execute function as a Ray remote function, the type of `results` is `RayStorage`,\n    when operand is executed, it will fetch dependencies from a Ray actor.\n    \"\"\"\n\n    _graph_execution_cls = GraphExecutionForRay\n\n    @classmethod\n    def handle(cls, op, results, mock=False):\n        method_name, mapper = ('execute', cls._op_runners) if not mock else \\\n            ('estimate_size', cls._op_size_estimators)\n        try:\n            runner = mapper[type(op)]\n        except KeyError:\n            runner = getattr(op, method_name)\n\n        # register a custom serializer for Mars operand\n        _register_ray_serializer(op)\n\n        try:\n            ray.wait([execute_on_ray.remote(runner, results, op)])\n        except NotImplementedError:\n            for op_cls in mapper.keys():\n                if isinstance(op, op_cls):\n                    mapper[type(op)] = mapper[op_cls]\n                    runner = mapper[op_cls]\n\n                    ray.wait(\n                        [execute_on_ray.remote(runner, results, op)])\n            raise KeyError(f'No handler found for op: {op}')\n\n    @classmethod\n    def _get_chunk_shape(cls, chunk_key, chunk_result):\n        assert isinstance(chunk_result, RayStorage)\n        return ray.get(chunk_result.meta_store.get_shape.remote(chunk_key))\n\n\nclass RaySession:\n    \"\"\"\n    Session to submit Mars job to Ray cluster.\n\n    If Ray is not initialized, kwargs will pass to initialize Ray.\n    \"\"\"\n    def __init__(self, **kwargs):\n        # as we cannot serialize fuse chunk for now,\n        # we just disable numexpr for ray executor\n        engine = kwargs.pop('engine', ['numpy', 'dataframe'])\n        if not ray.is_initialized():\n            ray.init(**kwargs)\n        self._session_id = uuid.uuid4()\n        self._executor = RayExecutor(engine=engine,\n                                     storage=RayStorage())\n\n    @property\n    def session_id(self):\n        return self._session_id\n\n    @property\n    def executor(self):\n        return self._executor\n\n    def fetch(self, *tileables, **kw):\n        return self._executor.fetch_tileables(tileables, **kw)\n\n    def fetch_log(self, tileables, offsets=None, sizes=None):  # pragma: no cover\n        raise NotImplementedError('`fetch_log` is not implemented for ray executor')\n\n    def run(self, *tileables, **kw):\n        \"\"\"\n        Parallelism equals to Ray cluster CPUs.\n        \"\"\"\n        if 'n_parallel' not in kw:  # pragma: no cover\n            kw['n_parallel'] = ray.cluster_resources()['CPU']\n        return self._executor.execute_tileables(tileables, **kw)\n\n    def _update_tileable_shape(self, tileable):\n        from ..optimizes.tileable_graph import tileable_optimized\n\n        new_nsplits = self._executor.get_tileable_nsplits(tileable)\n        tiled = get_tiled(tileable, mapping=tileable_optimized)\n        for t in (tileable, tiled):\n            t._update_shape(tuple(sum(nsplit) for nsplit in new_nsplits))\n        tiled.nsplits = new_nsplits\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, *_):\n        self._executor = None\n", "path": "mars/ray/core.py"}]}