[core] runtime context resource ids getter

python/ray/_private/worker.py

@@ -860,7 +860,7 @@ def get_gpu_ids():
     return assigned_ids
 
 
-@Deprecated
+@Deprecated(message="Use ray.get_runtime_context().assigned_resources instead.")
 def get_resource_ids():
     """Get the IDs of the resources that are available to the worker.
 

python/ray/runtime_context.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Dict, Any
 
 import ray._private.worker
 from ray._private.client_mode_hook import client_mode_hook
@@ -16,7 +17,7 @@ def __init__(self, worker):
         assert worker is not None
         self.worker = worker
 
-    def get(self):
+    def get(self) -> Dict[str, Any]:
         """Get a dictionary of the current context.
 
         Returns:
@@ -160,6 +161,18 @@ def should_capture_child_tasks_in_placement_group(self):
         """
         return self.worker.should_capture_child_tasks_in_placement_group
 
+    @property
+    def assigned_resources(self):
+        """Get the assigned resources to this worker.
+
+        Returns:
+            A dictionary mapping the name of a resource to a list of pairs, where
+            each pair consists of the ID of a resource and the fraction of that
+            resource reserved for this worker.
+        """
+        self.worker.check_connected()
+        return self.worker.core_worker.resource_ids()
+
     def get_runtime_env_string(self):
         """Get the runtime env string used for the current driver or worker.
 

python/ray/tests/test_basic_3.py

@@ -71,7 +71,7 @@ def g():
     def f(block, accepted_resources):
         true_resources = {
             resource: value[0][1]
-            for resource, value in ray._private.worker.get_resource_ids().items()
+            for resource, value in ray.get_runtime_context().assigned_resources.items()
         }
         if block:
             ray.get(g.remote())

python/ray/tests/test_placement_group.py

@@ -370,7 +370,7 @@ def test_placement_group_actor_resource_ids(ray_start_cluster, connect_to_client
     @ray.remote(num_cpus=1)
     class F:
         def f(self):
-            return ray._private.worker.get_resource_ids()
+            return ray.get_runtime_context().assigned_resources
 
     cluster = ray_start_cluster
     num_nodes = 1
@@ -391,7 +391,7 @@ def f(self):
 def test_placement_group_task_resource_ids(ray_start_cluster, connect_to_client):
     @ray.remote(num_cpus=1)
     def f():
-        return ray._private.worker.get_resource_ids()
+        return ray.get_runtime_context().assigned_resources
 
     cluster = ray_start_cluster
     num_nodes = 1
@@ -423,7 +423,7 @@ def f():
 def test_placement_group_hang(ray_start_cluster, connect_to_client):
     @ray.remote(num_cpus=1)
     def f():
-        return ray._private.worker.get_resource_ids()
+        return ray.get_runtime_context().assigned_resources
 
     cluster = ray_start_cluster
     num_nodes = 1