We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] Ray storage case failed for Ray 1.3.0
<!--
Thank you for your contribution!

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

**Describe the bug**

Ray storage case failed for Ray 1.3.0 .

Error message:

```
__________________________ test_base_operations[ray] ___________________________

storage_context = <mars.storage.ray.RayStorage object at 0x7fe06827c280>

    @pytest.mark.asyncio
    @require_lib
    async def test_base_operations(storage_context):
        storage = storage_context
    
        data1 = np.random.rand(10, 10)
        put_info1 = await storage.put(data1)
        get_data1 = await storage.get(put_info1.object_id)
        np.testing.assert_array_equal(data1, get_data1)
    
        info1 = await storage.object_info(put_info1.object_id)
        assert info1.size == put_info1.size
    
        data2 = pd.DataFrame({'col1': np.arange(10),
                              'col2': [f'str{i}' for i in range(10)],
                              'col3': np.random.rand(10)},)
        put_info2 = await storage.put(data2)
        get_data2 = await storage.get(put_info2.object_id)
        pd.testing.assert_frame_equal(data2, get_data2)
    
        info2 = await storage.object_info(put_info2.object_id)
        assert info2.size == put_info2.size
    
        # FIXME: remove when list functionality is ready for vineyard.
        if not isinstance(storage, (VineyardStorage, SharedMemoryStorage, RayStorage)):
            num = len(await storage.list())
            assert num == 2
            await storage.delete(info2.object_id)
    
        # test SparseMatrix
        s1 = sps.csr_matrix([[1, 0, 1], [0, 0, 1]])
        s = SparseNDArray(s1)
        put_info3 = await storage.put(s)
        get_data3 = await storage.get(put_info3.object_id)
        assert isinstance(get_data3, SparseMatrix)
        np.testing.assert_array_equal(get_data3.toarray(), s1.A)
        np.testing.assert_array_equal(get_data3.todense(), s1.A)
    
        # test writer and reader
        t = np.random.random(10)
        b = dataserializer.dumps(t)
        async with await storage.open_writer(size=len(b)) as writer:
            split = len(b) // 2
            await writer.write(b[:split])
>           await writer.write(b[split:])

mars/storage/tests/test_libs.py:174: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
mars/lib/aio/file.py:56: in __aexit__
    await self.close()
mars/lib/aio/base.py:25: in method
    return await self._loop.run_in_executor(self._executor, func)
../../../miniconda/envs/test/lib/python3.8/concurrent/futures/thread.py:57: in run
    result = self.fn(*self.args, **self.kwargs)
mars/storage/core.py:110: in close
    self._write_close()
mars/storage/ray.py:76: in _write_close
    worker.core_worker.put_file_like_object(metadata, self._buffer.tell(), self._buffer,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

>   ???
E   TypeError: put_file_like_object() takes exactly 5 positional arguments (4 given)

python/ray/_raylet.pyx:1032: TypeError
```

--- 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/storage/ray.py`
Content:
```
1 from typing import Any, Dict, List, Tuple
2 from ..lib import sparse
3 from ..utils import lazy_import, implements, register_ray_serializer
4 from .base import StorageBackend, StorageLevel, ObjectInfo, register_storage_backend
5 from .core import BufferWrappedFileObject, StorageFileObject
6 
7 ray = lazy_import("ray")
8 
9 
10 # TODO(fyrestone): make the SparseMatrix pickleable.
11 
12 def _mars_sparse_matrix_serializer(value):
13     return [value.shape, value.spmatrix]
14 
15 
16 def _mars_sparse_matrix_deserializer(obj) -> sparse.SparseNDArray:
17     shape, spmatrix = obj
18     return sparse.matrix.SparseMatrix(spmatrix, shape=shape)
19 
20 
21 def _register_sparse_matrix_serializer():
22     # register a custom serializer for Mars SparseMatrix
23     register_ray_serializer(sparse.matrix.SparseMatrix,
24                             serializer=_mars_sparse_matrix_serializer,
25                             deserializer=_mars_sparse_matrix_deserializer)
26 
27 
28 class RayFileLikeObject:
29     def __init__(self):
30         self._buffers = []
31         self._size = 0
32 
33     def write(self, content: bytes):
34         self._buffers.append(content)
35         self._size += len(content)
36 
37     def readinto(self, buffer):
38         read_bytes = 0
39         for b in self._buffers:
40             read_pos = read_bytes + len(b)
41             buffer[read_bytes:read_pos] = b
42             read_bytes = read_pos
43         return read_bytes
44 
45     def close(self):
46         self._buffers.clear()
47         self._size = 0
48 
49     def tell(self):
50         return self._size
51 
52 
53 class RayFileObject(BufferWrappedFileObject):
54     def __init__(self, object_id: Any, mode: str):
55         self._object_id = object_id
56         super().__init__(mode, size=0)
57 
58     def _write_init(self):
59         self._buffer = RayFileLikeObject()
60 
61     def _read_init(self):
62         self._buffer = ray.get(self._object_id)
63         self._mv = memoryview(self._buffer)
64         self._size = len(self._buffer)
65 
66     def write(self, content: bytes):
67         if not self._initialized:
68             self._write_init()
69             self._initialized = True
70 
71         return self._buffer.write(content)
72 
73     def _write_close(self):
74         worker = ray.worker.global_worker
75         metadata = ray.ray_constants.OBJECT_METADATA_TYPE_RAW
76         worker.core_worker.put_file_like_object(metadata, self._buffer.tell(), self._buffer,
77                                                 self._object_id)
78 
79     def _read_close(self):
80         pass
81 
82 
83 @register_storage_backend
84 class RayStorage(StorageBackend):
85     name = 'ray'
86 
87     def __init__(self, *args, **kwargs):
88         pass
89 
90     @classmethod
91     @implements(StorageBackend.setup)
92     async def setup(cls, **kwargs) -> Tuple[Dict, Dict]:
93         ray.init(ignore_reinit_error=True)
94         _register_sparse_matrix_serializer()
95         return dict(), dict()
96 
97     @staticmethod
98     @implements(StorageBackend.teardown)
99     async def teardown(**kwargs):
100         pass
101 
102     @property
103     @implements(StorageBackend.level)
104     def level(self) -> StorageLevel:
105         # TODO(fyrestone): return StorageLevel.MEMORY & StorageLevel.DISK
106         # if object spilling is available.
107         return StorageLevel.MEMORY | StorageLevel.REMOTE
108 
109     @implements(StorageBackend.get)
110     async def get(self, object_id, **kwargs) -> object:
111         if kwargs:  # pragma: no cover
112             raise NotImplementedError(f'Got unsupported args: {",".join(kwargs)}')
113         return await object_id
114 
115     @implements(StorageBackend.put)
116     async def put(self, obj, importance=0) -> ObjectInfo:
117         object_id = ray.put(obj)
118         # We can't get the serialized bytes length from ray.put
119         return ObjectInfo(object_id=object_id)
120 
121     @implements(StorageBackend.delete)
122     async def delete(self, object_id):
123         ray.internal.free(object_id)
124 
125     @implements(StorageBackend.object_info)
126     async def object_info(self, object_id) -> ObjectInfo:
127         # The performance of obtaining the object size is poor.
128         return ObjectInfo(object_id=object_id)
129 
130     @implements(StorageBackend.open_writer)
131     async def open_writer(self, size=None) -> StorageFileObject:
132         new_id = ray.ObjectRef.from_random()
133         ray_writer = RayFileObject(new_id, mode='w')
134         return StorageFileObject(ray_writer, object_id=new_id)
135 
136     @implements(StorageBackend.open_reader)
137     async def open_reader(self, object_id) -> StorageFileObject:
138         ray_reader = RayFileObject(object_id, mode='r')
139         return StorageFileObject(ray_reader, object_id=object_id)
140 
141     @implements(StorageBackend.list)
142     async def list(self) -> List:
143         raise NotImplementedError("Ray storage does not support list")
144 
145     @implements(StorageBackend.prefetch)
146     async def prefetch(self, object_id):
147         pass
148
```
--- 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/storage/ray.py b/mars/storage/ray.py
--- a/mars/storage/ray.py
+++ b/mars/storage/ray.py
@@ -73,8 +73,12 @@
     def _write_close(self):
         worker = ray.worker.global_worker
         metadata = ray.ray_constants.OBJECT_METADATA_TYPE_RAW
-        worker.core_worker.put_file_like_object(metadata, self._buffer.tell(), self._buffer,
-                                                self._object_id)
+        args = [metadata, self._buffer.tell(), self._buffer, self._object_id]
+        try:
+            worker.core_worker.put_file_like_object(*args)
+        except TypeError:
+            args.append(None)  # owner_address for ray >= 1.3.0
+            worker.core_worker.put_file_like_object(*args)
 
     def _read_close(self):
         pass

{"golden_diff": "diff --git a/mars/storage/ray.py b/mars/storage/ray.py\n--- a/mars/storage/ray.py\n+++ b/mars/storage/ray.py\n@@ -73,8 +73,12 @@\n     def _write_close(self):\n         worker = ray.worker.global_worker\n         metadata = ray.ray_constants.OBJECT_METADATA_TYPE_RAW\n-        worker.core_worker.put_file_like_object(metadata, self._buffer.tell(), self._buffer,\n-                                                self._object_id)\n+        args = [metadata, self._buffer.tell(), self._buffer, self._object_id]\n+        try:\n+            worker.core_worker.put_file_like_object(*args)\n+        except TypeError:\n+            args.append(None)  # owner_address for ray >= 1.3.0\n+            worker.core_worker.put_file_like_object(*args)\n \n     def _read_close(self):\n         pass\n", "issue": "[BUG] Ray storage case failed for Ray 1.3.0\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\nRay storage case failed for Ray 1.3.0 .\r\n\r\nError message:\r\n\r\n```\r\n__________________________ test_base_operations[ray] ___________________________\r\n\r\nstorage_context = <mars.storage.ray.RayStorage object at 0x7fe06827c280>\r\n\r\n    @pytest.mark.asyncio\r\n    @require_lib\r\n    async def test_base_operations(storage_context):\r\n        storage = storage_context\r\n    \r\n        data1 = np.random.rand(10, 10)\r\n        put_info1 = await storage.put(data1)\r\n        get_data1 = await storage.get(put_info1.object_id)\r\n        np.testing.assert_array_equal(data1, get_data1)\r\n    \r\n        info1 = await storage.object_info(put_info1.object_id)\r\n        assert info1.size == put_info1.size\r\n    \r\n        data2 = pd.DataFrame({'col1': np.arange(10),\r\n                              'col2': [f'str{i}' for i in range(10)],\r\n                              'col3': np.random.rand(10)},)\r\n        put_info2 = await storage.put(data2)\r\n        get_data2 = await storage.get(put_info2.object_id)\r\n        pd.testing.assert_frame_equal(data2, get_data2)\r\n    \r\n        info2 = await storage.object_info(put_info2.object_id)\r\n        assert info2.size == put_info2.size\r\n    \r\n        # FIXME: remove when list functionality is ready for vineyard.\r\n        if not isinstance(storage, (VineyardStorage, SharedMemoryStorage, RayStorage)):\r\n            num = len(await storage.list())\r\n            assert num == 2\r\n            await storage.delete(info2.object_id)\r\n    \r\n        # test SparseMatrix\r\n        s1 = sps.csr_matrix([[1, 0, 1], [0, 0, 1]])\r\n        s = SparseNDArray(s1)\r\n        put_info3 = await storage.put(s)\r\n        get_data3 = await storage.get(put_info3.object_id)\r\n        assert isinstance(get_data3, SparseMatrix)\r\n        np.testing.assert_array_equal(get_data3.toarray(), s1.A)\r\n        np.testing.assert_array_equal(get_data3.todense(), s1.A)\r\n    \r\n        # test writer and reader\r\n        t = np.random.random(10)\r\n        b = dataserializer.dumps(t)\r\n        async with await storage.open_writer(size=len(b)) as writer:\r\n            split = len(b) // 2\r\n            await writer.write(b[:split])\r\n>           await writer.write(b[split:])\r\n\r\nmars/storage/tests/test_libs.py:174: \r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\nmars/lib/aio/file.py:56: in __aexit__\r\n    await self.close()\r\nmars/lib/aio/base.py:25: in method\r\n    return await self._loop.run_in_executor(self._executor, func)\r\n../../../miniconda/envs/test/lib/python3.8/concurrent/futures/thread.py:57: in run\r\n    result = self.fn(*self.args, **self.kwargs)\r\nmars/storage/core.py:110: in close\r\n    self._write_close()\r\nmars/storage/ray.py:76: in _write_close\r\n    worker.core_worker.put_file_like_object(metadata, self._buffer.tell(), self._buffer,\r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\n\r\n>   ???\r\nE   TypeError: put_file_like_object() takes exactly 5 positional arguments (4 given)\r\n\r\npython/ray/_raylet.pyx:1032: TypeError\r\n```\n", "before_files": [{"content": "from typing import Any, Dict, List, Tuple\nfrom ..lib import sparse\nfrom ..utils import lazy_import, implements, register_ray_serializer\nfrom .base import StorageBackend, StorageLevel, ObjectInfo, register_storage_backend\nfrom .core import BufferWrappedFileObject, StorageFileObject\n\nray = lazy_import(\"ray\")\n\n\n# TODO(fyrestone): make the SparseMatrix pickleable.\n\ndef _mars_sparse_matrix_serializer(value):\n    return [value.shape, value.spmatrix]\n\n\ndef _mars_sparse_matrix_deserializer(obj) -> sparse.SparseNDArray:\n    shape, spmatrix = obj\n    return sparse.matrix.SparseMatrix(spmatrix, shape=shape)\n\n\ndef _register_sparse_matrix_serializer():\n    # register a custom serializer for Mars SparseMatrix\n    register_ray_serializer(sparse.matrix.SparseMatrix,\n                            serializer=_mars_sparse_matrix_serializer,\n                            deserializer=_mars_sparse_matrix_deserializer)\n\n\nclass RayFileLikeObject:\n    def __init__(self):\n        self._buffers = []\n        self._size = 0\n\n    def write(self, content: bytes):\n        self._buffers.append(content)\n        self._size += len(content)\n\n    def readinto(self, buffer):\n        read_bytes = 0\n        for b in self._buffers:\n            read_pos = read_bytes + len(b)\n            buffer[read_bytes:read_pos] = b\n            read_bytes = read_pos\n        return read_bytes\n\n    def close(self):\n        self._buffers.clear()\n        self._size = 0\n\n    def tell(self):\n        return self._size\n\n\nclass RayFileObject(BufferWrappedFileObject):\n    def __init__(self, object_id: Any, mode: str):\n        self._object_id = object_id\n        super().__init__(mode, size=0)\n\n    def _write_init(self):\n        self._buffer = RayFileLikeObject()\n\n    def _read_init(self):\n        self._buffer = ray.get(self._object_id)\n        self._mv = memoryview(self._buffer)\n        self._size = len(self._buffer)\n\n    def write(self, content: bytes):\n        if not self._initialized:\n            self._write_init()\n            self._initialized = True\n\n        return self._buffer.write(content)\n\n    def _write_close(self):\n        worker = ray.worker.global_worker\n        metadata = ray.ray_constants.OBJECT_METADATA_TYPE_RAW\n        worker.core_worker.put_file_like_object(metadata, self._buffer.tell(), self._buffer,\n                                                self._object_id)\n\n    def _read_close(self):\n        pass\n\n\n@register_storage_backend\nclass RayStorage(StorageBackend):\n    name = 'ray'\n\n    def __init__(self, *args, **kwargs):\n        pass\n\n    @classmethod\n    @implements(StorageBackend.setup)\n    async def setup(cls, **kwargs) -> Tuple[Dict, Dict]:\n        ray.init(ignore_reinit_error=True)\n        _register_sparse_matrix_serializer()\n        return dict(), dict()\n\n    @staticmethod\n    @implements(StorageBackend.teardown)\n    async def teardown(**kwargs):\n        pass\n\n    @property\n    @implements(StorageBackend.level)\n    def level(self) -> StorageLevel:\n        # TODO(fyrestone): return StorageLevel.MEMORY & StorageLevel.DISK\n        # if object spilling is available.\n        return StorageLevel.MEMORY | StorageLevel.REMOTE\n\n    @implements(StorageBackend.get)\n    async def get(self, object_id, **kwargs) -> object:\n        if kwargs:  # pragma: no cover\n            raise NotImplementedError(f'Got unsupported args: {\",\".join(kwargs)}')\n        return await object_id\n\n    @implements(StorageBackend.put)\n    async def put(self, obj, importance=0) -> ObjectInfo:\n        object_id = ray.put(obj)\n        # We can't get the serialized bytes length from ray.put\n        return ObjectInfo(object_id=object_id)\n\n    @implements(StorageBackend.delete)\n    async def delete(self, object_id):\n        ray.internal.free(object_id)\n\n    @implements(StorageBackend.object_info)\n    async def object_info(self, object_id) -> ObjectInfo:\n        # The performance of obtaining the object size is poor.\n        return ObjectInfo(object_id=object_id)\n\n    @implements(StorageBackend.open_writer)\n    async def open_writer(self, size=None) -> StorageFileObject:\n        new_id = ray.ObjectRef.from_random()\n        ray_writer = RayFileObject(new_id, mode='w')\n        return StorageFileObject(ray_writer, object_id=new_id)\n\n    @implements(StorageBackend.open_reader)\n    async def open_reader(self, object_id) -> StorageFileObject:\n        ray_reader = RayFileObject(object_id, mode='r')\n        return StorageFileObject(ray_reader, object_id=object_id)\n\n    @implements(StorageBackend.list)\n    async def list(self) -> List:\n        raise NotImplementedError(\"Ray storage does not support list\")\n\n    @implements(StorageBackend.prefetch)\n    async def prefetch(self, object_id):\n        pass\n", "path": "mars/storage/ray.py"}], "after_files": [{"content": "from typing import Any, Dict, List, Tuple\nfrom ..lib import sparse\nfrom ..utils import lazy_import, implements, register_ray_serializer\nfrom .base import StorageBackend, StorageLevel, ObjectInfo, register_storage_backend\nfrom .core import BufferWrappedFileObject, StorageFileObject\n\nray = lazy_import(\"ray\")\n\n\n# TODO(fyrestone): make the SparseMatrix pickleable.\n\ndef _mars_sparse_matrix_serializer(value):\n    return [value.shape, value.spmatrix]\n\n\ndef _mars_sparse_matrix_deserializer(obj) -> sparse.SparseNDArray:\n    shape, spmatrix = obj\n    return sparse.matrix.SparseMatrix(spmatrix, shape=shape)\n\n\ndef _register_sparse_matrix_serializer():\n    # register a custom serializer for Mars SparseMatrix\n    register_ray_serializer(sparse.matrix.SparseMatrix,\n                            serializer=_mars_sparse_matrix_serializer,\n                            deserializer=_mars_sparse_matrix_deserializer)\n\n\nclass RayFileLikeObject:\n    def __init__(self):\n        self._buffers = []\n        self._size = 0\n\n    def write(self, content: bytes):\n        self._buffers.append(content)\n        self._size += len(content)\n\n    def readinto(self, buffer):\n        read_bytes = 0\n        for b in self._buffers:\n            read_pos = read_bytes + len(b)\n            buffer[read_bytes:read_pos] = b\n            read_bytes = read_pos\n        return read_bytes\n\n    def close(self):\n        self._buffers.clear()\n        self._size = 0\n\n    def tell(self):\n        return self._size\n\n\nclass RayFileObject(BufferWrappedFileObject):\n    def __init__(self, object_id: Any, mode: str):\n        self._object_id = object_id\n        super().__init__(mode, size=0)\n\n    def _write_init(self):\n        self._buffer = RayFileLikeObject()\n\n    def _read_init(self):\n        self._buffer = ray.get(self._object_id)\n        self._mv = memoryview(self._buffer)\n        self._size = len(self._buffer)\n\n    def write(self, content: bytes):\n        if not self._initialized:\n            self._write_init()\n            self._initialized = True\n\n        return self._buffer.write(content)\n\n    def _write_close(self):\n        worker = ray.worker.global_worker\n        metadata = ray.ray_constants.OBJECT_METADATA_TYPE_RAW\n        args = [metadata, self._buffer.tell(), self._buffer, self._object_id]\n        try:\n            worker.core_worker.put_file_like_object(*args)\n        except TypeError:\n            args.append(None)  # owner_address for ray >= 1.3.0\n            worker.core_worker.put_file_like_object(*args)\n\n    def _read_close(self):\n        pass\n\n\n@register_storage_backend\nclass RayStorage(StorageBackend):\n    name = 'ray'\n\n    def __init__(self, *args, **kwargs):\n        pass\n\n    @classmethod\n    @implements(StorageBackend.setup)\n    async def setup(cls, **kwargs) -> Tuple[Dict, Dict]:\n        ray.init(ignore_reinit_error=True)\n        _register_sparse_matrix_serializer()\n        return dict(), dict()\n\n    @staticmethod\n    @implements(StorageBackend.teardown)\n    async def teardown(**kwargs):\n        pass\n\n    @property\n    @implements(StorageBackend.level)\n    def level(self) -> StorageLevel:\n        # TODO(fyrestone): return StorageLevel.MEMORY & StorageLevel.DISK\n        # if object spilling is available.\n        return StorageLevel.MEMORY | StorageLevel.REMOTE\n\n    @implements(StorageBackend.get)\n    async def get(self, object_id, **kwargs) -> object:\n        if kwargs:  # pragma: no cover\n            raise NotImplementedError(f'Got unsupported args: {\",\".join(kwargs)}')\n        return await object_id\n\n    @implements(StorageBackend.put)\n    async def put(self, obj, importance=0) -> ObjectInfo:\n        object_id = ray.put(obj)\n        # We can't get the serialized bytes length from ray.put\n        return ObjectInfo(object_id=object_id)\n\n    @implements(StorageBackend.delete)\n    async def delete(self, object_id):\n        ray.internal.free(object_id)\n\n    @implements(StorageBackend.object_info)\n    async def object_info(self, object_id) -> ObjectInfo:\n        # The performance of obtaining the object size is poor.\n        return ObjectInfo(object_id=object_id)\n\n    @implements(StorageBackend.open_writer)\n    async def open_writer(self, size=None) -> StorageFileObject:\n        new_id = ray.ObjectRef.from_random()\n        ray_writer = RayFileObject(new_id, mode='w')\n        return StorageFileObject(ray_writer, object_id=new_id)\n\n    @implements(StorageBackend.open_reader)\n    async def open_reader(self, object_id) -> StorageFileObject:\n        ray_reader = RayFileObject(object_id, mode='r')\n        return StorageFileObject(ray_reader, object_id=object_id)\n\n    @implements(StorageBackend.list)\n    async def list(self) -> List:\n        raise NotImplementedError(\"Ray storage does not support list\")\n\n    @implements(StorageBackend.prefetch)\n    async def prefetch(self, object_id):\n        pass\n", "path": "mars/storage/ray.py"}]}