(refactor): extract gpu & ray utils.

Author: FightingZhen

roll/distributed/executor/cluster.py

@@ -17,7 +17,6 @@
     collect_all_to_all,
     dispatch_one_to_all,
 )
-from roll.platforms import current_platform
 from roll.utils.constants import RAY_NAMESPACE
 from roll.distributed.scheduler.resource_manager import ResourceManager
 from roll.utils.import_utils import safe_import_class
@@ -123,25 +122,14 @@ def _create_workers(self):
 
             runtime_env = RuntimeEnv(env_vars=env_vars)
             self.worker_config.resource_placement_groups = pgs
-            if current_platform.ray_device_key == "GPU":
-                worker = self.worker_cls.options(
-                    scheduling_strategy=PlacementGroupSchedulingStrategy(placement_group=deploy_pg["placement_group"]),
-                    name=worker_name,
-                    namespace=RAY_NAMESPACE,
-                    runtime_env=runtime_env,
-                    num_cpus=0.01,
-                    num_gpus=0.01 if self.worker_config.device_mapping else 0,
-                ).remote(worker_config=self.worker_config)
-            else:
-                worker = self.worker_cls.options(
-                    scheduling_strategy=PlacementGroupSchedulingStrategy(placement_group=deploy_pg["placement_group"]),
-                    name=worker_name,
-                    namespace=RAY_NAMESPACE,
-                    runtime_env=runtime_env,
-                    num_cpus=0.01,
-                    num_gpus=0,
-                    resources={current_platform.ray_device_key: 0.01 if self.worker_config.device_mapping else 0},
-                ).remote(worker_config=self.worker_config)
+            worker = self.worker_cls.options(
+                scheduling_strategy=PlacementGroupSchedulingStrategy(placement_group=deploy_pg["placement_group"]),
+                name=worker_name,
+                namespace=RAY_NAMESPACE,
+                runtime_env=runtime_env,
+                num_cpus=0.01,
+                num_gpus=0.01 if self.worker_config.device_mapping else 0,
+            ).remote(worker_config=self.worker_config)
             self.workers.append(worker)
             if rank == 0:
                 self.master_addr, self.master_port = ray.get(worker.get_master_addr_and_port.remote())