(self,
method: str,
args: tuple = (),
kwargs: Optional[dict] = None,
non_block: bool = False,
unique_reply_rank: Optional[int] = None)
| 193 | |
| 194 | @unwrap_ray_errors() |
| 195 | def collective_rpc(self, |
| 196 | method: str, |
| 197 | args: tuple = (), |
| 198 | kwargs: Optional[dict] = None, |
| 199 | non_block: bool = False, |
| 200 | unique_reply_rank: Optional[int] = None) -> list[Any]: |
| 201 | workers = (self.workers[unique_reply_rank], |
| 202 | ) if unique_reply_rank is not None else self.workers |
| 203 | kwargs = kwargs or {} |
| 204 | |
| 205 | refs = [] |
| 206 | for w in workers: |
| 207 | try: |
| 208 | refs.append(getattr(w, method).remote(*args, **kwargs)) |
| 209 | except AttributeError: |
| 210 | # Here worker is the RayWorkerWrapper. |
| 211 | # For extended worker methods, we need to use call_worker_method since |
| 212 | # Ray actor doesn't work with __getattr__ delegation. |
| 213 | refs.append(w.call_worker_method.remote(method, *args, |
| 214 | **kwargs)) |
| 215 | return refs if non_block else ray.get(refs) |
| 216 | |
| 217 | @unwrap_ray_errors() |
| 218 | async def collective_rpc_async( |
no test coverage detected