From a699f6a4d82a7103a3c78f56ceb1ec0027e25e9a Mon Sep 17 00:00:00 2001
From: Alex Wu <itswu.alex@gmail.com>
Date: Sun, 6 Sep 2020 20:56:48 -0700
Subject: [PATCH] [Core] Fix override memory and object_store_memory in
 decorator (#10563)

---
 python/ray/tests/test_basic.py | 24 ++++++++++++++++++++++++
 python/ray/utils.py            |  7 ++++---
 rllib/evaluation/worker_set.py |  7 ++++---
 3 files changed, 32 insertions(+), 6 deletions(-)

diff --git a/python/ray/tests/test_basic.py b/python/ray/tests/test_basic.py
index 63837c8a2..375f2938a 100644
--- a/python/ray/tests/test_basic.py
+++ b/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.
diff --git a/python/ray/utils.py b/python/ray/utils.py
index f5071a7f9..e7323fc9a 100644
--- a/python/ray/utils.py
+++ b/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)
diff --git a/rllib/evaluation/worker_set.py b/rllib/evaluation/worker_set.py
index 0f0f4f057..8d278f2d4 100644
--- a/rllib/evaluation/worker_set.py
+++ b/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