[Fix] replace ray.get to await in async func and add timeout in ray.get #1397
Conversation
YanhuiDua
changed the title
|
|
||
| # TODO: Move DEVICE to `xtuner.utils.device` | ||
| PG_READY_TIMEOUT = 30 | ||
| TRAINER_RAY_GET_TIMEOUT = 5 * 3600 # 5 hour |
There was a problem hiding this comment.
Both PG_READY_TIMEOUT and TRAINER_RAY_GET_TIMEOUT should be configurable by environment variable. Consider create a env.py to manager cross-file env that have the same meaning.
There was a problem hiding this comment.
下个pr汇合下所有的env，统一管理，需要讨论下env.py放在什么目录下
| self.logging_replaybuffer_state() | ||
| self.logger.info(ray.get(self.env_controller.get_rollout_stats.remote())) # type: ignore[attr-defined] | ||
| replay_buffer_stats = await self.replay_buffer.print.remote() # type: ignore[attr-defined] | ||
| rollout_stats = await self.env_controller.get_rollout_stats.remote() # type: ignore[attr-defined] |
There was a problem hiding this comment.
Why not using asyncio.wait here? Since it's at the end of this method, not inside a task while loop.
There was a problem hiding this comment.
Ray 的 ObjectRef（如 actor.method.remote() 返回的对象）不能直接用于 asyncio.wait，需要经过一些转换，see at: https://docs.ray.io/en/latest/ray-core/actors/async_api.html#objectrefs-as-asyncio-futures
| await asyncio.wait_for(asyncio.gather(*waiting_tasks, return_exceptions=True), timeout=10) | ||
|
|
||
| self.logger.info(ray.get(self.env_controller.get_rollout_stats.remote())) # type: ignore[attr-defined] | ||
| rollout_stats = await self.env_controller.get_rollout_stats.remote() # type: ignore[attr-defined] |
There was a problem hiding this comment.
Why not using asyncio.wait here? Since it's at the end of this method, not inside a task while loop.
No description provided.