[Core] Fix override memory and object_store_memory in decorator (#10563)

python/ray/tests/test_basic.py

@@ -341,6 +341,30 @@ def test_function_descriptor():
     assert d.get(python_descriptor2) == 123
 
 
+def test_ray_options(shutdown_only):
+    @ray.remote(
+        num_cpus=2, num_gpus=3, memory=150 * 2**20, resources={"custom1": 1})
+    def foo():
+        return ray.available_resources()
+
+    ray.init(num_cpus=10, num_gpus=10, resources={"custom1": 2})
+
+    without_options = ray.get(foo.remote())
+    with_options = ray.get(
+        foo.options(
+            num_cpus=3,
+            num_gpus=4,
+            memory=50 * 2**20,
+            resources={
+                "custom1": 0.5
+            }).remote())
+
+    to_check = ["CPU", "GPU", "memory", "custom1"]
+    for key in to_check:
+        assert without_options[key] != with_options[key]
+    assert without_options != with_options
+
+
 def test_nested_functions(ray_start_shared_local_modes):
     # Make sure that remote functions can use other values that are defined
     # after the remote function but before the first function invocation.

python/ray/utils.py

@@ -365,9 +365,10 @@ def resources_from_resource_arguments(
     elif default_num_gpus is not None:
         resources["GPU"] = default_num_gpus
 
-    memory = default_memory or runtime_memory
-    object_store_memory = (default_object_store_memory
-                           or runtime_object_store_memory)
+    # Order of arguments matter for short circuiting.
+    memory = runtime_memory or default_memory
+    object_store_memory = (runtime_object_store_memory
+                           or default_object_store_memory)
     if memory is not None:
         resources["memory"] = ray_constants.to_memory_units(
             memory, round_up=True)

rllib/evaluation/worker_set.py

@@ -98,9 +98,10 @@ class WorkerSet:
         remote_args = {
             "num_cpus": self._remote_config["num_cpus_per_worker"],
             "num_gpus": self._remote_config["num_gpus_per_worker"],
-            "memory": self._remote_config["memory_per_worker"],
-            "object_store_memory": self._remote_config[
-                "object_store_memory_per_worker"],
+            # memory=0 is an error, but memory=None means no limits.
+            "memory": self._remote_config["memory_per_worker"] or None,
+            "object_store_memory": self.
+            _remote_config["object_store_memory_per_worker"] or None,
             "resources": self._remote_config["custom_resources_per_worker"],
         }
         cls = RolloutWorker.as_remote(**remote_args).remote