We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] There maybe duplicate `WorkerMetaAPI` creation in `TaskStageProcessor`
**Describe the bug**

We can see there is a lot of duplicate `WorkerMetaAPI` creation in the method `_update_result_meta` of `mars/services/task/execution/mars/stage.py` if there is too many chunk result and their bands are same:
```python
for tileable in tile_context.values():
    chunks = [c.data for c in tileable.chunks]
    for c, params_fields in zip(chunks, self._get_params_fields(tileable)):
        address = chunk_to_result[c].meta["bands"][0][0]
        meta_api = await WorkerMetaAPI.create(session_id, address)
        call = meta_api.get_chunk_meta.delay(c.key, fields=params_fields)
        worker_meta_api_to_chunk_delays[meta_api][c] = calll
```
There are 3639 chunk results and 200 workers in my job and it costs 1414s to create worker meta apis.
Maybe we can cache the worker meta apis to avoid duplicate creation.

**To Reproduce**

1. Your Python version: python-3.7.9
2. The version of Mars you use: master


--- 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/services/meta/api/oscar.py`
Content:
```
1 # Copyright 1999-2021 Alibaba Group Holding Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 
16 from typing import Dict, List, Any
17 
18 from .... import oscar as mo
19 from ....core import ChunkType
20 from ....core.operand import Fuse
21 from ....lib.aio import alru_cache
22 from ....typing import BandType
23 from ....utils import get_chunk_params
24 from ..core import get_meta_type
25 from ..store import AbstractMetaStore
26 from ..supervisor.core import MetaStoreManagerActor, MetaStoreActor
27 from ..worker.core import WorkerMetaStoreManagerActor
28 from .core import AbstractMetaAPI
29 
30 
31 class BaseMetaAPI(AbstractMetaAPI):
32     def __init__(self, session_id: str, meta_store: mo.ActorRefType[AbstractMetaStore]):
33         # make sure all meta types registered
34         from .. import metas
35 
36         del metas
37 
38         self._session_id = session_id
39         self._meta_store = meta_store
40 
41     @mo.extensible
42     async def set_tileable_meta(
43         self, tileable, memory_size: int = None, store_size: int = None, **extra
44     ):
45         from ....dataframe.core import (
46             DATAFRAME_TYPE,
47             DATAFRAME_GROUPBY_TYPE,
48             SERIES_GROUPBY_TYPE,
49         )
50 
51         params = tileable.params.copy()
52         if isinstance(
53             tileable, (DATAFRAME_TYPE, DATAFRAME_GROUPBY_TYPE, SERIES_GROUPBY_TYPE)
54         ):
55             # dataframe needs some special process for now
56             del params["columns_value"]
57             del params["dtypes"]
58             params.pop("key_dtypes", None)
59             params["dtypes_value"] = tileable.dtypes_value
60         params["nsplits"] = tileable.nsplits
61         params.update(extra)
62         meta = get_meta_type(type(tileable))(
63             object_id=tileable.key,
64             **params,
65             memory_size=memory_size,
66             store_size=store_size
67         )
68         return await self._meta_store.set_meta(tileable.key, meta)
69 
70     @mo.extensible
71     async def get_tileable_meta(
72         self, object_id: str, fields: List[str] = None
73     ) -> Dict[str, Any]:
74         return await self._meta_store.get_meta(object_id, fields=fields)
75 
76     @mo.extensible
77     async def del_tileable_meta(self, object_id: str):
78         return await self._meta_store.del_meta(object_id)
79 
80     @classmethod
81     def _extract_chunk_meta(
82         cls,
83         chunk: ChunkType,
84         memory_size: int = None,
85         store_size: int = None,
86         bands: List[BandType] = None,
87         fields: List[str] = None,
88         exclude_fields: List[str] = None,
89         **extra
90     ):
91         if isinstance(chunk.op, Fuse):
92             # fuse op
93             chunk = chunk.chunk
94         params = get_chunk_params(chunk)
95         chunk_key = extra.pop("chunk_key", chunk.key)
96         object_ref = extra.pop("object_ref", None)
97         params.update(extra)
98 
99         if object_ref:
100             object_refs = (
101                 [object_ref] if not isinstance(object_ref, list) else object_ref
102             )
103         else:
104             object_refs = []
105 
106         if fields is not None:
107             fields = set(fields)
108             params = {k: v for k, v in params.items() if k in fields}
109         elif exclude_fields is not None:
110             exclude_fields = set(exclude_fields)
111             params = {k: v for k, v in params.items() if k not in exclude_fields}
112 
113         return get_meta_type(type(chunk))(
114             object_id=chunk_key,
115             **params,
116             bands=bands,
117             memory_size=memory_size,
118             store_size=store_size,
119             object_refs=object_refs
120         )
121 
122     @mo.extensible
123     async def set_chunk_meta(
124         self,
125         chunk: ChunkType,
126         memory_size: int = None,
127         store_size: int = None,
128         bands: List[BandType] = None,
129         fields: List[str] = None,
130         exclude_fields: List[str] = None,
131         **extra
132     ):
133         """
134         Parameters
135         ----------
136         chunk: ChunkType
137             chunk to set meta
138         memory_size: int
139             memory size for chunk data
140         store_size: int
141             serialized size for chunk data
142         bands:
143             chunk data bands
144         fields: list
145             fields to include in meta
146         exclude_fields: list
147             fields to exclude in meta
148         extra
149 
150         Returns
151         -------
152 
153         """
154         meta = self._extract_chunk_meta(
155             chunk,
156             memory_size=memory_size,
157             store_size=store_size,
158             bands=bands,
159             fields=fields,
160             exclude_fields=exclude_fields,
161             **extra
162         )
163         return await self._meta_store.set_meta(meta.object_id, meta)
164 
165     @set_chunk_meta.batch
166     async def batch_set_chunk_meta(self, args_list, kwargs_list):
167         set_chunk_metas = []
168         for args, kwargs in zip(args_list, kwargs_list):
169             meta = self._extract_chunk_meta(*args, **kwargs)
170             set_chunk_metas.append(
171                 self._meta_store.set_meta.delay(meta.object_id, meta)
172             )
173         return await self._meta_store.set_meta.batch(*set_chunk_metas)
174 
175     @mo.extensible
176     async def get_chunk_meta(
177         self, object_id: str, fields: List[str] = None, error="raise"
178     ):
179         return await self._meta_store.get_meta(object_id, fields=fields, error=error)
180 
181     @get_chunk_meta.batch
182     async def batch_get_chunk_meta(self, args_list, kwargs_list):
183         get_chunk_metas = []
184         for args, kwargs in zip(args_list, kwargs_list):
185             get_chunk_metas.append(self._meta_store.get_meta.delay(*args, **kwargs))
186         return await self._meta_store.get_meta.batch(*get_chunk_metas)
187 
188     @mo.extensible
189     async def del_chunk_meta(self, object_id: str):
190         """
191         Parameters
192         ----------
193         object_id: str
194             chunk id
195         """
196         return await self._meta_store.del_meta(object_id)
197 
198     @del_chunk_meta.batch
199     async def batch_del_chunk_meta(self, args_list, kwargs_list):
200         del_chunk_metas = []
201         for args, kwargs in zip(args_list, kwargs_list):
202             del_chunk_metas.append(self._meta_store.del_meta.delay(*args, **kwargs))
203         return await self._meta_store.del_meta.batch(*del_chunk_metas)
204 
205     @mo.extensible
206     async def add_chunk_bands(self, object_id: str, bands: List[BandType]):
207         return await self._meta_store.add_chunk_bands(object_id, bands)
208 
209     @add_chunk_bands.batch
210     async def batch_add_chunk_bands(self, args_list, kwargs_list):
211         add_chunk_bands_tasks = []
212         for args, kwargs in zip(args_list, kwargs_list):
213             add_chunk_bands_tasks.append(
214                 self._meta_store.add_chunk_bands.delay(*args, **kwargs)
215             )
216         return await self._meta_store.add_chunk_bands.batch(*add_chunk_bands_tasks)
217 
218     @mo.extensible
219     async def remove_chunk_bands(self, object_id: str, bands: List[BandType]):
220         return await self._meta_store.remove_chunk_bands(object_id, bands)
221 
222     @remove_chunk_bands.batch
223     async def batch_remove_chunk_bands(self, args_list, kwargs_list):
224         remove_chunk_bands_tasks = []
225         for args, kwargs in zip(args_list, kwargs_list):
226             remove_chunk_bands_tasks.append(
227                 self._meta_store.remove_chunk_bands.delay(*args, **kwargs)
228             )
229         return await self._meta_store.remove_chunk_bands.batch(
230             *remove_chunk_bands_tasks
231         )
232 
233     @mo.extensible
234     async def get_band_chunks(self, band: BandType) -> List[str]:
235         return await self._meta_store.get_band_chunks(band)
236 
237 
238 class MetaAPI(BaseMetaAPI):
239     @classmethod
240     @alru_cache(cache_exceptions=False)
241     async def create(cls, session_id: str, address: str) -> "MetaAPI":
242         """
243         Create Meta API.
244 
245         Parameters
246         ----------
247         session_id : str
248             Session ID.
249         address : str
250             Supervisor address.
251 
252         Returns
253         -------
254         meta_api
255             Meta api.
256         """
257         meta_store_ref = await mo.actor_ref(address, MetaStoreActor.gen_uid(session_id))
258 
259         return MetaAPI(session_id, meta_store_ref)
260 
261 
262 class MockMetaAPI(MetaAPI):
263     @classmethod
264     async def create(cls, session_id: str, address: str) -> "MetaAPI":
265         # create an Actor for mock
266         try:
267             meta_store_manager_ref = await mo.create_actor(
268                 MetaStoreManagerActor,
269                 "dict",
270                 dict(),
271                 address=address,
272                 uid=MetaStoreManagerActor.default_uid(),
273             )
274         except mo.ActorAlreadyExist:
275             # ignore if actor exists
276             meta_store_manager_ref = await mo.actor_ref(
277                 MetaStoreManagerActor,
278                 address=address,
279                 uid=MetaStoreManagerActor.default_uid(),
280             )
281         try:
282             await meta_store_manager_ref.new_session_meta_store(session_id)
283         except mo.ActorAlreadyExist:
284             pass
285         return await super().create(session_id=session_id, address=address)
286 
287 
288 class WorkerMetaAPI(BaseMetaAPI):
289     @classmethod
290     @alru_cache(cache_exceptions=False)
291     async def create(cls, session_id: str, address: str) -> "WorkerMetaAPI":
292         """
293         Create worker meta API.
294 
295         Parameters
296         ----------
297         session_id : str
298             Session ID.
299         address : str
300             Worker address.
301 
302         Returns
303         -------
304         meta_api
305             Worker meta api.
306         """
307         worker_meta_store_manager_ref = await mo.actor_ref(
308             uid=WorkerMetaStoreManagerActor.default_uid(), address=address
309         )
310         worker_meta_store_ref = (
311             await worker_meta_store_manager_ref.new_session_meta_store(session_id)
312         )
313         return WorkerMetaAPI(session_id, worker_meta_store_ref)
314 
315 
316 class MockWorkerMetaAPI(WorkerMetaAPI):
317     @classmethod
318     async def create(cls, session_id: str, address: str) -> "WorkerMetaAPI":
319         # create an Actor for mock
320         try:
321             await mo.create_actor(
322                 WorkerMetaStoreManagerActor,
323                 "dict",
324                 dict(),
325                 address=address,
326                 uid=WorkerMetaStoreManagerActor.default_uid(),
327             )
328         except mo.ActorAlreadyExist:
329             # ignore if actor exists
330             await mo.actor_ref(
331                 WorkerMetaStoreManagerActor,
332                 address=address,
333                 uid=WorkerMetaStoreManagerActor.default_uid(),
334             )
335         return await super().create(session_id, address)
336
```
--- 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/services/meta/api/oscar.py b/mars/services/meta/api/oscar.py
--- a/mars/services/meta/api/oscar.py
+++ b/mars/services/meta/api/oscar.py
@@ -237,7 +237,7 @@
 
 class MetaAPI(BaseMetaAPI):
     @classmethod
-    @alru_cache(cache_exceptions=False)
+    @alru_cache(maxsize=1024, cache_exceptions=False)
     async def create(cls, session_id: str, address: str) -> "MetaAPI":
         """
         Create Meta API.

{"golden_diff": "diff --git a/mars/services/meta/api/oscar.py b/mars/services/meta/api/oscar.py\n--- a/mars/services/meta/api/oscar.py\n+++ b/mars/services/meta/api/oscar.py\n@@ -237,7 +237,7 @@\n \n class MetaAPI(BaseMetaAPI):\n     @classmethod\n-    @alru_cache(cache_exceptions=False)\n+    @alru_cache(maxsize=1024, cache_exceptions=False)\n     async def create(cls, session_id: str, address: str) -> \"MetaAPI\":\n         \"\"\"\n         Create Meta API.\n", "issue": "[BUG] There maybe duplicate `WorkerMetaAPI` creation in `TaskStageProcessor`\n**Describe the bug**\r\n\r\nWe can see there is a lot of duplicate `WorkerMetaAPI` creation in the method `_update_result_meta` of `mars/services/task/execution/mars/stage.py` if there is too many chunk result and their bands are same:\r\n```python\r\nfor tileable in tile_context.values():\r\n    chunks = [c.data for c in tileable.chunks]\r\n    for c, params_fields in zip(chunks, self._get_params_fields(tileable)):\r\n        address = chunk_to_result[c].meta[\"bands\"][0][0]\r\n        meta_api = await WorkerMetaAPI.create(session_id, address)\r\n        call = meta_api.get_chunk_meta.delay(c.key, fields=params_fields)\r\n        worker_meta_api_to_chunk_delays[meta_api][c] = calll\r\n```\r\nThere are 3639 chunk results and 200 workers in my job and it costs 1414s to create worker meta apis.\r\nMaybe we can cache the worker meta apis to avoid duplicate creation.\r\n\r\n**To Reproduce**\r\n\r\n1. Your Python version: python-3.7.9\r\n2. The version of Mars you use: master\r\n\n", "before_files": [{"content": "# Copyright 1999-2021 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\n\nfrom typing import Dict, List, Any\n\nfrom .... import oscar as mo\nfrom ....core import ChunkType\nfrom ....core.operand import Fuse\nfrom ....lib.aio import alru_cache\nfrom ....typing import BandType\nfrom ....utils import get_chunk_params\nfrom ..core import get_meta_type\nfrom ..store import AbstractMetaStore\nfrom ..supervisor.core import MetaStoreManagerActor, MetaStoreActor\nfrom ..worker.core import WorkerMetaStoreManagerActor\nfrom .core import AbstractMetaAPI\n\n\nclass BaseMetaAPI(AbstractMetaAPI):\n    def __init__(self, session_id: str, meta_store: mo.ActorRefType[AbstractMetaStore]):\n        # make sure all meta types registered\n        from .. import metas\n\n        del metas\n\n        self._session_id = session_id\n        self._meta_store = meta_store\n\n    @mo.extensible\n    async def set_tileable_meta(\n        self, tileable, memory_size: int = None, store_size: int = None, **extra\n    ):\n        from ....dataframe.core import (\n            DATAFRAME_TYPE,\n            DATAFRAME_GROUPBY_TYPE,\n            SERIES_GROUPBY_TYPE,\n        )\n\n        params = tileable.params.copy()\n        if isinstance(\n            tileable, (DATAFRAME_TYPE, DATAFRAME_GROUPBY_TYPE, SERIES_GROUPBY_TYPE)\n        ):\n            # dataframe needs some special process for now\n            del params[\"columns_value\"]\n            del params[\"dtypes\"]\n            params.pop(\"key_dtypes\", None)\n            params[\"dtypes_value\"] = tileable.dtypes_value\n        params[\"nsplits\"] = tileable.nsplits\n        params.update(extra)\n        meta = get_meta_type(type(tileable))(\n            object_id=tileable.key,\n            **params,\n            memory_size=memory_size,\n            store_size=store_size\n        )\n        return await self._meta_store.set_meta(tileable.key, meta)\n\n    @mo.extensible\n    async def get_tileable_meta(\n        self, object_id: str, fields: List[str] = None\n    ) -> Dict[str, Any]:\n        return await self._meta_store.get_meta(object_id, fields=fields)\n\n    @mo.extensible\n    async def del_tileable_meta(self, object_id: str):\n        return await self._meta_store.del_meta(object_id)\n\n    @classmethod\n    def _extract_chunk_meta(\n        cls,\n        chunk: ChunkType,\n        memory_size: int = None,\n        store_size: int = None,\n        bands: List[BandType] = None,\n        fields: List[str] = None,\n        exclude_fields: List[str] = None,\n        **extra\n    ):\n        if isinstance(chunk.op, Fuse):\n            # fuse op\n            chunk = chunk.chunk\n        params = get_chunk_params(chunk)\n        chunk_key = extra.pop(\"chunk_key\", chunk.key)\n        object_ref = extra.pop(\"object_ref\", None)\n        params.update(extra)\n\n        if object_ref:\n            object_refs = (\n                [object_ref] if not isinstance(object_ref, list) else object_ref\n            )\n        else:\n            object_refs = []\n\n        if fields is not None:\n            fields = set(fields)\n            params = {k: v for k, v in params.items() if k in fields}\n        elif exclude_fields is not None:\n            exclude_fields = set(exclude_fields)\n            params = {k: v for k, v in params.items() if k not in exclude_fields}\n\n        return get_meta_type(type(chunk))(\n            object_id=chunk_key,\n            **params,\n            bands=bands,\n            memory_size=memory_size,\n            store_size=store_size,\n            object_refs=object_refs\n        )\n\n    @mo.extensible\n    async def set_chunk_meta(\n        self,\n        chunk: ChunkType,\n        memory_size: int = None,\n        store_size: int = None,\n        bands: List[BandType] = None,\n        fields: List[str] = None,\n        exclude_fields: List[str] = None,\n        **extra\n    ):\n        \"\"\"\n        Parameters\n        ----------\n        chunk: ChunkType\n            chunk to set meta\n        memory_size: int\n            memory size for chunk data\n        store_size: int\n            serialized size for chunk data\n        bands:\n            chunk data bands\n        fields: list\n            fields to include in meta\n        exclude_fields: list\n            fields to exclude in meta\n        extra\n\n        Returns\n        -------\n\n        \"\"\"\n        meta = self._extract_chunk_meta(\n            chunk,\n            memory_size=memory_size,\n            store_size=store_size,\n            bands=bands,\n            fields=fields,\n            exclude_fields=exclude_fields,\n            **extra\n        )\n        return await self._meta_store.set_meta(meta.object_id, meta)\n\n    @set_chunk_meta.batch\n    async def batch_set_chunk_meta(self, args_list, kwargs_list):\n        set_chunk_metas = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            meta = self._extract_chunk_meta(*args, **kwargs)\n            set_chunk_metas.append(\n                self._meta_store.set_meta.delay(meta.object_id, meta)\n            )\n        return await self._meta_store.set_meta.batch(*set_chunk_metas)\n\n    @mo.extensible\n    async def get_chunk_meta(\n        self, object_id: str, fields: List[str] = None, error=\"raise\"\n    ):\n        return await self._meta_store.get_meta(object_id, fields=fields, error=error)\n\n    @get_chunk_meta.batch\n    async def batch_get_chunk_meta(self, args_list, kwargs_list):\n        get_chunk_metas = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            get_chunk_metas.append(self._meta_store.get_meta.delay(*args, **kwargs))\n        return await self._meta_store.get_meta.batch(*get_chunk_metas)\n\n    @mo.extensible\n    async def del_chunk_meta(self, object_id: str):\n        \"\"\"\n        Parameters\n        ----------\n        object_id: str\n            chunk id\n        \"\"\"\n        return await self._meta_store.del_meta(object_id)\n\n    @del_chunk_meta.batch\n    async def batch_del_chunk_meta(self, args_list, kwargs_list):\n        del_chunk_metas = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            del_chunk_metas.append(self._meta_store.del_meta.delay(*args, **kwargs))\n        return await self._meta_store.del_meta.batch(*del_chunk_metas)\n\n    @mo.extensible\n    async def add_chunk_bands(self, object_id: str, bands: List[BandType]):\n        return await self._meta_store.add_chunk_bands(object_id, bands)\n\n    @add_chunk_bands.batch\n    async def batch_add_chunk_bands(self, args_list, kwargs_list):\n        add_chunk_bands_tasks = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            add_chunk_bands_tasks.append(\n                self._meta_store.add_chunk_bands.delay(*args, **kwargs)\n            )\n        return await self._meta_store.add_chunk_bands.batch(*add_chunk_bands_tasks)\n\n    @mo.extensible\n    async def remove_chunk_bands(self, object_id: str, bands: List[BandType]):\n        return await self._meta_store.remove_chunk_bands(object_id, bands)\n\n    @remove_chunk_bands.batch\n    async def batch_remove_chunk_bands(self, args_list, kwargs_list):\n        remove_chunk_bands_tasks = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            remove_chunk_bands_tasks.append(\n                self._meta_store.remove_chunk_bands.delay(*args, **kwargs)\n            )\n        return await self._meta_store.remove_chunk_bands.batch(\n            *remove_chunk_bands_tasks\n        )\n\n    @mo.extensible\n    async def get_band_chunks(self, band: BandType) -> List[str]:\n        return await self._meta_store.get_band_chunks(band)\n\n\nclass MetaAPI(BaseMetaAPI):\n    @classmethod\n    @alru_cache(cache_exceptions=False)\n    async def create(cls, session_id: str, address: str) -> \"MetaAPI\":\n        \"\"\"\n        Create Meta API.\n\n        Parameters\n        ----------\n        session_id : str\n            Session ID.\n        address : str\n            Supervisor address.\n\n        Returns\n        -------\n        meta_api\n            Meta api.\n        \"\"\"\n        meta_store_ref = await mo.actor_ref(address, MetaStoreActor.gen_uid(session_id))\n\n        return MetaAPI(session_id, meta_store_ref)\n\n\nclass MockMetaAPI(MetaAPI):\n    @classmethod\n    async def create(cls, session_id: str, address: str) -> \"MetaAPI\":\n        # create an Actor for mock\n        try:\n            meta_store_manager_ref = await mo.create_actor(\n                MetaStoreManagerActor,\n                \"dict\",\n                dict(),\n                address=address,\n                uid=MetaStoreManagerActor.default_uid(),\n            )\n        except mo.ActorAlreadyExist:\n            # ignore if actor exists\n            meta_store_manager_ref = await mo.actor_ref(\n                MetaStoreManagerActor,\n                address=address,\n                uid=MetaStoreManagerActor.default_uid(),\n            )\n        try:\n            await meta_store_manager_ref.new_session_meta_store(session_id)\n        except mo.ActorAlreadyExist:\n            pass\n        return await super().create(session_id=session_id, address=address)\n\n\nclass WorkerMetaAPI(BaseMetaAPI):\n    @classmethod\n    @alru_cache(cache_exceptions=False)\n    async def create(cls, session_id: str, address: str) -> \"WorkerMetaAPI\":\n        \"\"\"\n        Create worker meta API.\n\n        Parameters\n        ----------\n        session_id : str\n            Session ID.\n        address : str\n            Worker address.\n\n        Returns\n        -------\n        meta_api\n            Worker meta api.\n        \"\"\"\n        worker_meta_store_manager_ref = await mo.actor_ref(\n            uid=WorkerMetaStoreManagerActor.default_uid(), address=address\n        )\n        worker_meta_store_ref = (\n            await worker_meta_store_manager_ref.new_session_meta_store(session_id)\n        )\n        return WorkerMetaAPI(session_id, worker_meta_store_ref)\n\n\nclass MockWorkerMetaAPI(WorkerMetaAPI):\n    @classmethod\n    async def create(cls, session_id: str, address: str) -> \"WorkerMetaAPI\":\n        # create an Actor for mock\n        try:\n            await mo.create_actor(\n                WorkerMetaStoreManagerActor,\n                \"dict\",\n                dict(),\n                address=address,\n                uid=WorkerMetaStoreManagerActor.default_uid(),\n            )\n        except mo.ActorAlreadyExist:\n            # ignore if actor exists\n            await mo.actor_ref(\n                WorkerMetaStoreManagerActor,\n                address=address,\n                uid=WorkerMetaStoreManagerActor.default_uid(),\n            )\n        return await super().create(session_id, address)\n", "path": "mars/services/meta/api/oscar.py"}], "after_files": [{"content": "# Copyright 1999-2021 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\n\nfrom typing import Dict, List, Any\n\nfrom .... import oscar as mo\nfrom ....core import ChunkType\nfrom ....core.operand import Fuse\nfrom ....lib.aio import alru_cache\nfrom ....typing import BandType\nfrom ....utils import get_chunk_params\nfrom ..core import get_meta_type\nfrom ..store import AbstractMetaStore\nfrom ..supervisor.core import MetaStoreManagerActor, MetaStoreActor\nfrom ..worker.core import WorkerMetaStoreManagerActor\nfrom .core import AbstractMetaAPI\n\n\nclass BaseMetaAPI(AbstractMetaAPI):\n    def __init__(self, session_id: str, meta_store: mo.ActorRefType[AbstractMetaStore]):\n        # make sure all meta types registered\n        from .. import metas\n\n        del metas\n\n        self._session_id = session_id\n        self._meta_store = meta_store\n\n    @mo.extensible\n    async def set_tileable_meta(\n        self, tileable, memory_size: int = None, store_size: int = None, **extra\n    ):\n        from ....dataframe.core import (\n            DATAFRAME_TYPE,\n            DATAFRAME_GROUPBY_TYPE,\n            SERIES_GROUPBY_TYPE,\n        )\n\n        params = tileable.params.copy()\n        if isinstance(\n            tileable, (DATAFRAME_TYPE, DATAFRAME_GROUPBY_TYPE, SERIES_GROUPBY_TYPE)\n        ):\n            # dataframe needs some special process for now\n            del params[\"columns_value\"]\n            del params[\"dtypes\"]\n            params.pop(\"key_dtypes\", None)\n            params[\"dtypes_value\"] = tileable.dtypes_value\n        params[\"nsplits\"] = tileable.nsplits\n        params.update(extra)\n        meta = get_meta_type(type(tileable))(\n            object_id=tileable.key,\n            **params,\n            memory_size=memory_size,\n            store_size=store_size\n        )\n        return await self._meta_store.set_meta(tileable.key, meta)\n\n    @mo.extensible\n    async def get_tileable_meta(\n        self, object_id: str, fields: List[str] = None\n    ) -> Dict[str, Any]:\n        return await self._meta_store.get_meta(object_id, fields=fields)\n\n    @mo.extensible\n    async def del_tileable_meta(self, object_id: str):\n        return await self._meta_store.del_meta(object_id)\n\n    @classmethod\n    def _extract_chunk_meta(\n        cls,\n        chunk: ChunkType,\n        memory_size: int = None,\n        store_size: int = None,\n        bands: List[BandType] = None,\n        fields: List[str] = None,\n        exclude_fields: List[str] = None,\n        **extra\n    ):\n        if isinstance(chunk.op, Fuse):\n            # fuse op\n            chunk = chunk.chunk\n        params = get_chunk_params(chunk)\n        chunk_key = extra.pop(\"chunk_key\", chunk.key)\n        object_ref = extra.pop(\"object_ref\", None)\n        params.update(extra)\n\n        if object_ref:\n            object_refs = (\n                [object_ref] if not isinstance(object_ref, list) else object_ref\n            )\n        else:\n            object_refs = []\n\n        if fields is not None:\n            fields = set(fields)\n            params = {k: v for k, v in params.items() if k in fields}\n        elif exclude_fields is not None:\n            exclude_fields = set(exclude_fields)\n            params = {k: v for k, v in params.items() if k not in exclude_fields}\n\n        return get_meta_type(type(chunk))(\n            object_id=chunk_key,\n            **params,\n            bands=bands,\n            memory_size=memory_size,\n            store_size=store_size,\n            object_refs=object_refs\n        )\n\n    @mo.extensible\n    async def set_chunk_meta(\n        self,\n        chunk: ChunkType,\n        memory_size: int = None,\n        store_size: int = None,\n        bands: List[BandType] = None,\n        fields: List[str] = None,\n        exclude_fields: List[str] = None,\n        **extra\n    ):\n        \"\"\"\n        Parameters\n        ----------\n        chunk: ChunkType\n            chunk to set meta\n        memory_size: int\n            memory size for chunk data\n        store_size: int\n            serialized size for chunk data\n        bands:\n            chunk data bands\n        fields: list\n            fields to include in meta\n        exclude_fields: list\n            fields to exclude in meta\n        extra\n\n        Returns\n        -------\n\n        \"\"\"\n        meta = self._extract_chunk_meta(\n            chunk,\n            memory_size=memory_size,\n            store_size=store_size,\n            bands=bands,\n            fields=fields,\n            exclude_fields=exclude_fields,\n            **extra\n        )\n        return await self._meta_store.set_meta(meta.object_id, meta)\n\n    @set_chunk_meta.batch\n    async def batch_set_chunk_meta(self, args_list, kwargs_list):\n        set_chunk_metas = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            meta = self._extract_chunk_meta(*args, **kwargs)\n            set_chunk_metas.append(\n                self._meta_store.set_meta.delay(meta.object_id, meta)\n            )\n        return await self._meta_store.set_meta.batch(*set_chunk_metas)\n\n    @mo.extensible\n    async def get_chunk_meta(\n        self, object_id: str, fields: List[str] = None, error=\"raise\"\n    ):\n        return await self._meta_store.get_meta(object_id, fields=fields, error=error)\n\n    @get_chunk_meta.batch\n    async def batch_get_chunk_meta(self, args_list, kwargs_list):\n        get_chunk_metas = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            get_chunk_metas.append(self._meta_store.get_meta.delay(*args, **kwargs))\n        return await self._meta_store.get_meta.batch(*get_chunk_metas)\n\n    @mo.extensible\n    async def del_chunk_meta(self, object_id: str):\n        \"\"\"\n        Parameters\n        ----------\n        object_id: str\n            chunk id\n        \"\"\"\n        return await self._meta_store.del_meta(object_id)\n\n    @del_chunk_meta.batch\n    async def batch_del_chunk_meta(self, args_list, kwargs_list):\n        del_chunk_metas = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            del_chunk_metas.append(self._meta_store.del_meta.delay(*args, **kwargs))\n        return await self._meta_store.del_meta.batch(*del_chunk_metas)\n\n    @mo.extensible\n    async def add_chunk_bands(self, object_id: str, bands: List[BandType]):\n        return await self._meta_store.add_chunk_bands(object_id, bands)\n\n    @add_chunk_bands.batch\n    async def batch_add_chunk_bands(self, args_list, kwargs_list):\n        add_chunk_bands_tasks = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            add_chunk_bands_tasks.append(\n                self._meta_store.add_chunk_bands.delay(*args, **kwargs)\n            )\n        return await self._meta_store.add_chunk_bands.batch(*add_chunk_bands_tasks)\n\n    @mo.extensible\n    async def remove_chunk_bands(self, object_id: str, bands: List[BandType]):\n        return await self._meta_store.remove_chunk_bands(object_id, bands)\n\n    @remove_chunk_bands.batch\n    async def batch_remove_chunk_bands(self, args_list, kwargs_list):\n        remove_chunk_bands_tasks = []\n        for args, kwargs in zip(args_list, kwargs_list):\n            remove_chunk_bands_tasks.append(\n                self._meta_store.remove_chunk_bands.delay(*args, **kwargs)\n            )\n        return await self._meta_store.remove_chunk_bands.batch(\n            *remove_chunk_bands_tasks\n        )\n\n    @mo.extensible\n    async def get_band_chunks(self, band: BandType) -> List[str]:\n        return await self._meta_store.get_band_chunks(band)\n\n\nclass MetaAPI(BaseMetaAPI):\n    @classmethod\n    @alru_cache(maxsize=1024, cache_exceptions=False)\n    async def create(cls, session_id: str, address: str) -> \"MetaAPI\":\n        \"\"\"\n        Create Meta API.\n\n        Parameters\n        ----------\n        session_id : str\n            Session ID.\n        address : str\n            Supervisor address.\n\n        Returns\n        -------\n        meta_api\n            Meta api.\n        \"\"\"\n        meta_store_ref = await mo.actor_ref(address, MetaStoreActor.gen_uid(session_id))\n\n        return MetaAPI(session_id, meta_store_ref)\n\n\nclass MockMetaAPI(MetaAPI):\n    @classmethod\n    async def create(cls, session_id: str, address: str) -> \"MetaAPI\":\n        # create an Actor for mock\n        try:\n            meta_store_manager_ref = await mo.create_actor(\n                MetaStoreManagerActor,\n                \"dict\",\n                dict(),\n                address=address,\n                uid=MetaStoreManagerActor.default_uid(),\n            )\n        except mo.ActorAlreadyExist:\n            # ignore if actor exists\n            meta_store_manager_ref = await mo.actor_ref(\n                MetaStoreManagerActor,\n                address=address,\n                uid=MetaStoreManagerActor.default_uid(),\n            )\n        try:\n            await meta_store_manager_ref.new_session_meta_store(session_id)\n        except mo.ActorAlreadyExist:\n            pass\n        return await super().create(session_id=session_id, address=address)\n\n\nclass WorkerMetaAPI(BaseMetaAPI):\n    @classmethod\n    @alru_cache(cache_exceptions=False)\n    async def create(cls, session_id: str, address: str) -> \"WorkerMetaAPI\":\n        \"\"\"\n        Create worker meta API.\n\n        Parameters\n        ----------\n        session_id : str\n            Session ID.\n        address : str\n            Worker address.\n\n        Returns\n        -------\n        meta_api\n            Worker meta api.\n        \"\"\"\n        worker_meta_store_manager_ref = await mo.actor_ref(\n            uid=WorkerMetaStoreManagerActor.default_uid(), address=address\n        )\n        worker_meta_store_ref = (\n            await worker_meta_store_manager_ref.new_session_meta_store(session_id)\n        )\n        return WorkerMetaAPI(session_id, worker_meta_store_ref)\n\n\nclass MockWorkerMetaAPI(WorkerMetaAPI):\n    @classmethod\n    async def create(cls, session_id: str, address: str) -> \"WorkerMetaAPI\":\n        # create an Actor for mock\n        try:\n            await mo.create_actor(\n                WorkerMetaStoreManagerActor,\n                \"dict\",\n                dict(),\n                address=address,\n                uid=WorkerMetaStoreManagerActor.default_uid(),\n            )\n        except mo.ActorAlreadyExist:\n            # ignore if actor exists\n            await mo.actor_ref(\n                WorkerMetaStoreManagerActor,\n                address=address,\n                uid=WorkerMetaStoreManagerActor.default_uid(),\n            )\n        return await super().create(session_id, address)\n", "path": "mars/services/meta/api/oscar.py"}]}