[core] Store Internal Config in GCS (#8921)

python/ray/node.py

@@ -92,6 +92,11 @@ class Node:
                 "The raylet IP address should only be different than the node "
                 "IP address when connecting to an existing raylet; i.e., when "
                 "head=False and connect_only=True.")
+        if ray_params._internal_config and len(
+                ray_params._internal_config) > 0 and (not head
+                                                      and not connect_only):
+            raise ValueError(
+                "Internal config parameters can only be set on the head node.")
 
         self._raylet_ip_address = raylet_ip_address
 
@@ -663,7 +668,8 @@ class Node:
             plasma_directory=self._ray_params.plasma_directory,
             huge_pages=self._ray_params.huge_pages,
             fate_share=self.kernel_fate_share,
-            socket_to_use=self.socket)
+            socket_to_use=self.socket,
+            head_node=self.head)
         assert ray_constants.PROCESS_TYPE_RAYLET not in self.all_processes
         self.all_processes[ray_constants.PROCESS_TYPE_RAYLET] = [process_info]
 

python/ray/services.py

@@ -1260,7 +1260,8 @@ def start_raylet(redis_address,
                  plasma_directory=None,
                  huge_pages=False,
                  fate_share=None,
-                 socket_to_use=None):
+                 socket_to_use=None,
+                 head_node=False):
     """Start a raylet, which is a combined local scheduler and object manager.
 
     Args:
@@ -1402,6 +1403,8 @@ def start_raylet(redis_address,
             command.append("--huge_pages")
     if socket_to_use:
         socket_to_use.close()
+    if head_node:
+        command.append("--head_node")
     process_info = start_ray_process(
         command,
         ray_constants.PROCESS_TYPE_RAYLET,

python/ray/tests/conftest.py

@@ -111,7 +111,9 @@ def _ray_start_cluster(**kwargs):
     init_kwargs.update(kwargs)
     cluster = Cluster()
     remote_nodes = []
-    for _ in range(num_nodes):
+    for i in range(num_nodes):
+        if i > 0 and "_internal_config" in init_kwargs:
+            del init_kwargs["_internal_config"]
         remote_nodes.append(cluster.add_node(**init_kwargs))
         # We assume driver will connect to the head (first node),
         # so ray init will be invoked if do_init is true
@@ -200,8 +202,7 @@ def two_node_cluster():
     cluster = ray.cluster_utils.Cluster(
         head_node_args={"_internal_config": internal_config})
     for _ in range(2):
-        remote_node = cluster.add_node(
-            num_cpus=1, _internal_config=internal_config)
+        remote_node = cluster.add_node(num_cpus=1)
     ray.init(address=cluster.address)
     yield cluster, remote_node
 

python/ray/tests/test_actor_failures.py

@@ -291,7 +291,7 @@ def test_actor_restart_on_node_failure(ray_start_cluster):
     ray.init(address=cluster.address)
 
     # Node to place the actor.
-    actor_node = cluster.add_node(num_cpus=1, _internal_config=config)
+    actor_node = cluster.add_node(num_cpus=1)
     cluster.wait_for_nodes()
 
     @ray.remote(num_cpus=1, max_restarts=1, max_task_retries=-1)
@@ -313,7 +313,7 @@ def test_actor_restart_on_node_failure(ray_start_cluster):
     results = [actor.increase.remote() for _ in range(100)]
     # Kill actor node, while the above task is still being executed.
     cluster.remove_node(actor_node)
-    cluster.add_node(num_cpus=1, _internal_config=config)
+    cluster.add_node(num_cpus=1)
     cluster.wait_for_nodes()
     # Check that none of the tasks failed and the actor is restarted.
     seq = list(range(1, 101))
@@ -442,7 +442,8 @@ def test_caller_task_reconstruction(ray_start_regular):
 @pytest.mark.parametrize(
     "ray_start_cluster_head", [
         generate_internal_config_map(
-            initial_reconstruction_timeout_milliseconds=1000)
+            initial_reconstruction_timeout_milliseconds=1000,
+            num_heartbeats_timeout=10)
     ],
     indirect=True)
 def test_multiple_actor_restart(ray_start_cluster_head):
@@ -454,14 +455,7 @@ def test_multiple_actor_restart(ray_start_cluster_head):
     num_actors_at_a_time = 3
     num_function_calls_at_a_time = 10
 
-    worker_nodes = [
-        cluster.add_node(
-            num_cpus=3,
-            _internal_config=json.dumps({
-                "initial_reconstruction_timeout_milliseconds": 200,
-                "num_heartbeats_timeout": 10,
-            })) for _ in range(num_nodes)
-    ]
+    worker_nodes = [cluster.add_node(num_cpus=3) for _ in range(num_nodes)]
 
     @ray.remote(max_restarts=-1, max_task_retries=-1)
     class SlowCounter:

python/ray/tests/test_actor_resources.py

@@ -242,7 +242,7 @@ def test_actor_multiple_gpus_from_multiple_tasks(ray_start_cluster):
             num_gpus=num_gpus_per_raylet,
             _internal_config=json.dumps({
                 "num_heartbeats_timeout": 1000
-            }))
+            } if i == 0 else {}))
     ray.init(address=cluster.address)
 
     @ray.remote

python/ray/tests/test_failure.py

@@ -842,15 +842,15 @@ def test_connect_with_disconnected_node(shutdown_only):
     info = relevant_errors(ray_constants.REMOVED_NODE_ERROR)
     assert len(info) == 0
     # This node is killed by SIGKILL, ray_monitor will mark it to dead.
-    dead_node = cluster.add_node(num_cpus=0, _internal_config=config)
+    dead_node = cluster.add_node(num_cpus=0)
     cluster.remove_node(dead_node, allow_graceful=False)
     wait_for_errors(ray_constants.REMOVED_NODE_ERROR, 1)
     # This node is killed by SIGKILL, ray_monitor will mark it to dead.
-    dead_node = cluster.add_node(num_cpus=0, _internal_config=config)
+    dead_node = cluster.add_node(num_cpus=0)
     cluster.remove_node(dead_node, allow_graceful=False)
     wait_for_errors(ray_constants.REMOVED_NODE_ERROR, 2)
     # This node is killed by SIGTERM, ray_monitor will not mark it again.
-    removing_node = cluster.add_node(num_cpus=0, _internal_config=config)
+    removing_node = cluster.add_node(num_cpus=0)
     cluster.remove_node(removing_node, allow_graceful=True)
     with pytest.raises(RayTestTimeoutException):
         wait_for_errors(ray_constants.REMOVED_NODE_ERROR, 3, timeout=2)

python/ray/tests/test_multi_node_2.py

@@ -32,8 +32,11 @@ def test_shutdown():
 
 
 @pytest.mark.parametrize(
-    "ray_start_cluster_head",
-    [generate_internal_config_map(num_heartbeats_timeout=20)],
+    "ray_start_cluster_head", [
+        generate_internal_config_map(
+            num_heartbeats_timeout=20,
+            initial_reconstruction_timeout_milliseconds=12345)
+    ],
     indirect=True)
 def test_internal_config(ray_start_cluster_head):
     """Checks that the internal configuration setting works.
@@ -41,12 +44,20 @@ def test_internal_config(ray_start_cluster_head):
     We set the cluster to timeout nodes after 2 seconds of no timeouts. We
     then remove a node, wait for 1 second to check that the cluster is out
     of sync, then wait another 2 seconds (giving 1 second of leeway) to check
-    that the client has timed out.
+    that the client has timed out. We also check to see if the config is set.
     """
     cluster = ray_start_cluster_head
     worker = cluster.add_node()
     cluster.wait_for_nodes()
 
+    @ray.remote
+    def f():
+        assert ray._config.initial_reconstruction_timeout_milliseconds(
+        ) == 12345
+        assert ray._config.num_heartbeats_timeout() == 20
+
+    ray.get([f.remote() for _ in range(5)])
+
     cluster.remove_node(worker, allow_graceful=False)
     time.sleep(1)
     assert ray.cluster_resources()["CPU"] == 2

python/ray/tests/test_object_manager.py

@@ -214,10 +214,7 @@ def test_object_transfer_retry(ray_start_cluster):
     object_store_memory = 150 * 1024 * 1024
     cluster.add_node(
         object_store_memory=object_store_memory, _internal_config=config)
-    cluster.add_node(
-        num_gpus=1,
-        object_store_memory=object_store_memory,
-        _internal_config=config)
+    cluster.add_node(num_gpus=1, object_store_memory=object_store_memory)
     ray.init(address=cluster.address)
 
     @ray.remote(num_gpus=1)

python/ray/tests/test_reconstruction.py

@@ -63,15 +63,9 @@ def test_reconstruction_cached_dependency(ray_start_cluster,
     ray.init(address=cluster.address)
     # Node to place the initial object.
     node_to_kill = cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node2": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node2": 1}, object_store_memory=10**8)
     cluster.wait_for_nodes()
 
     @ray.remote(max_retries=0)
@@ -92,10 +86,7 @@ def test_reconstruction_cached_dependency(ray_start_cluster,
 
     cluster.remove_node(node_to_kill, allow_graceful=False)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
     assert wait_for_condition(
         lambda: not all(node["Alive"] for node in ray.nodes()), timeout=10)
 
@@ -125,15 +116,9 @@ def test_basic_reconstruction(ray_start_cluster, reconstruction_enabled):
     ray.init(address=cluster.address)
     # Node to place the initial object.
     node_to_kill = cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node2": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node2": 1}, object_store_memory=10**8)
     cluster.wait_for_nodes()
 
     @ray.remote(max_retries=1 if reconstruction_enabled else 0)
@@ -149,10 +134,7 @@ def test_basic_reconstruction(ray_start_cluster, reconstruction_enabled):
 
     cluster.remove_node(node_to_kill, allow_graceful=False)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
 
     if reconstruction_enabled:
         ray.get(dependent_task.remote(obj))
@@ -177,15 +159,9 @@ def test_basic_reconstruction_put(ray_start_cluster, reconstruction_enabled):
     ray.init(address=cluster.address)
     # Node to place the initial object.
     node_to_kill = cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node2": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node2": 1}, object_store_memory=10**8)
     cluster.wait_for_nodes()
 
     @ray.remote(max_retries=1 if reconstruction_enabled else 0)
@@ -203,10 +179,7 @@ def test_basic_reconstruction_put(ray_start_cluster, reconstruction_enabled):
 
     cluster.remove_node(node_to_kill, allow_graceful=False)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
 
     for _ in range(20):
         ray.put(np.zeros(10**7, dtype=np.uint8))
@@ -232,15 +205,9 @@ def test_multiple_downstream_tasks(ray_start_cluster, reconstruction_enabled):
     ray.init(address=cluster.address)
     # Node to place the initial object.
     node_to_kill = cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node2": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node2": 1}, object_store_memory=10**8)
     cluster.wait_for_nodes()
 
     @ray.remote(max_retries=1 if reconstruction_enabled else 0)
@@ -262,10 +229,7 @@ def test_multiple_downstream_tasks(ray_start_cluster, reconstruction_enabled):
 
     cluster.remove_node(node_to_kill, allow_graceful=False)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
 
     if reconstruction_enabled:
         for obj in downstream:
@@ -294,8 +258,7 @@ def test_reconstruction_chain(ray_start_cluster, reconstruction_enabled):
     cluster.add_node(
         num_cpus=0, _internal_config=config, object_store_memory=10**8)
     ray.init(address=cluster.address)
-    node_to_kill = cluster.add_node(
-        num_cpus=1, object_store_memory=10**8, _internal_config=config)
+    node_to_kill = cluster.add_node(num_cpus=1, object_store_memory=10**8)
     cluster.wait_for_nodes()
 
     @ray.remote(max_retries=1 if reconstruction_enabled else 0)
@@ -316,8 +279,7 @@ def test_reconstruction_chain(ray_start_cluster, reconstruction_enabled):
     ray.get(dependent_task.remote(obj))
 
     cluster.remove_node(node_to_kill, allow_graceful=False)
-    cluster.add_node(
-        num_cpus=1, object_store_memory=10**8, _internal_config=config)
+    cluster.add_node(num_cpus=1, object_store_memory=10**8)
 
     if reconstruction_enabled:
         ray.get(dependent_task.remote(obj))
@@ -343,15 +305,9 @@ def test_reconstruction_stress(ray_start_cluster):
     ray.init(address=cluster.address)
     # Node to place the initial object.
     node_to_kill = cluster.add_node(
-        num_cpus=1,
-        resources={"node1": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
     cluster.add_node(
-        num_cpus=1,
-        resources={"node2": 1},
-        object_store_memory=10**8,
-        _internal_config=config)
+        num_cpus=1, resources={"node2": 1}, object_store_memory=10**8)
     cluster.wait_for_nodes()
 
     @ray.remote
@@ -379,10 +335,7 @@ def test_reconstruction_stress(ray_start_cluster):
 
         cluster.remove_node(node_to_kill, allow_graceful=False)
         node_to_kill = cluster.add_node(
-            num_cpus=1,
-            resources={"node1": 1},
-            object_store_memory=10**8,
-            _internal_config=config)
+            num_cpus=1, resources={"node1": 1}, object_store_memory=10**8)
 
         i = 0
         while outputs:

python/ray/tests/test_stress_failure.py

@@ -29,11 +29,7 @@ def ray_start_reconstruction(request):
         })
     for i in range(num_nodes - 1):
         cluster.add_node(
-            num_cpus=1,
-            object_store_memory=plasma_store_memory // num_nodes,
-            _internal_config=json.dumps({
-                "initial_reconstruction_timeout_milliseconds": 200
-            }))
+            num_cpus=1, object_store_memory=plasma_store_memory // num_nodes)
     ray.init(address=cluster.address)
 
     yield plasma_store_memory, num_nodes, cluster

python/ray/worker.py

@@ -770,6 +770,9 @@ def init(address=None,
         if _internal_config is not None and len(_internal_config) != 0:
             raise ValueError("When connecting to an existing cluster, "
                              "_internal_config must not be provided.")
+        if lru_evict:
+            raise ValueError("When connecting to an existing cluster, "
+                             "lru_evict must not be provided.")
 
         # In this case, we only need to connect the node.
         ray_params = ray.parameter.RayParams(