[core][2/N] Remove redis_max_memory from Ray core (#51059)

This is a follow up of #51023.

---------

Signed-off-by: kaihsun <kaihsun@anyscale.com>

python/ray/_private/node.py

@@ -582,7 +582,6 @@ def merge_resources(env_dict, params_dict):
                     else object_store_memory
                 ),
                 resources,
-                self._ray_params.redis_max_memory,
             ).resolve(is_head=self.head, node_ip_address=self.node_ip_address)
         return self._resource_spec
 

python/ray/_private/parameter.py

@@ -35,10 +35,6 @@ class RayParams:
         memory: Total available memory for workers requesting memory.
         object_store_memory: The amount of memory (in bytes) to start the
             object store with.
-        redis_max_memory: The max amount of memory (in bytes) to allow redis
-            to use, or None for no limit. Once the limit is exceeded, redis
-            will start LRU eviction of entries. This only applies to the
-            sharded redis tables (task and object tables).
         object_manager_port int: The port to use for the object manager.
         node_manager_port: The port to use for the node manager.
         gcs_server_port: The port to use for the GCS server.
@@ -147,7 +143,6 @@ def __init__(
         labels: Optional[Dict[str, str]] = None,
         memory: Optional[float] = None,
         object_store_memory: Optional[float] = None,
-        redis_max_memory: Optional[float] = None,
         redis_port: Optional[int] = None,
         redis_shard_ports: Optional[List[int]] = None,
         object_manager_port: Optional[int] = None,
@@ -208,7 +203,6 @@ def __init__(
         self.memory = memory
         self.object_store_memory = object_store_memory
         self.resources = resources
-        self.redis_max_memory = redis_max_memory
         self.redis_port = redis_port
         self.redis_shard_ports = redis_shard_ports
         self.object_manager_port = object_manager_port

python/ray/_private/ray_constants.py

@@ -87,11 +87,6 @@ def env_set_by_user(key):
 # TODO(swang): Ideally this should be pulled directly from the
 # config in case the user overrides it.
 DEFAULT_MAX_DIRECT_CALL_OBJECT_SIZE = 100 * 1024
-# The default maximum number of bytes that the non-primary Redis shards are
-# allowed to use unless overridden by the user.
-DEFAULT_REDIS_MAX_MEMORY_BYTES = 10**10
-# The smallest cap on the memory used by Redis that we allow.
-REDIS_MINIMUM_MEMORY_BYTES = 10**7
 # Above this number of bytes, raise an error by default unless the user sets
 # RAY_ALLOW_SLOW_STORAGE=1. This avoids swapping with large object stores.
 REQUIRE_SHM_SIZE_THRESHOLD = 10**10

python/ray/_private/resource_spec.py

@@ -25,7 +25,6 @@ class ResourceSpec(
             "memory",
             "object_store_memory",
             "resources",
-            "redis_max_memory",
         ],
     )
 ):
@@ -43,11 +42,6 @@ class ResourceSpec(
             Note that when calling to_resource_dict(), this will be scaled down
             by 30% to account for the global plasma LRU reserve.
         resources: The custom resources allocated for this raylet.
-        redis_max_memory: The max amount of memory (in bytes) to allow each
-            redis shard to use. Once the limit is exceeded, redis will start
-            LRU eviction of entries. This only applies to the sharded redis
-            tables (task, object, and profile tables). By default, this is
-            capped at 10GB but can be set higher.
     """
 
     def __new__(
@@ -57,7 +51,6 @@ def __new__(
         memory=None,
         object_store_memory=None,
         resources=None,
-        redis_max_memory=None,
     ):
         return super(ResourceSpec, cls).__new__(
             cls,
@@ -66,7 +59,6 @@ def __new__(
             memory,
             object_store_memory,
             resources,
-            redis_max_memory,
         )
 
     def resolved(self):
@@ -272,27 +264,9 @@ def resolve(self, is_head: bool, node_ip_address: Optional[str] = None):
                 )
                 object_store_memory = object_store_memory_cap
 
-        redis_max_memory = self.redis_max_memory
-        if redis_max_memory is None:
-            redis_max_memory = min(
-                ray_constants.DEFAULT_REDIS_MAX_MEMORY_BYTES,
-                max(int(avail_memory * 0.1), ray_constants.REDIS_MINIMUM_MEMORY_BYTES),
-            )
-        if redis_max_memory < ray_constants.REDIS_MINIMUM_MEMORY_BYTES:
-            raise ValueError(
-                "Attempting to cap Redis memory usage at {} bytes, "
-                "but the minimum allowed is {} bytes.".format(
-                    redis_max_memory, ray_constants.REDIS_MINIMUM_MEMORY_BYTES
-                )
-            )
-
         memory = self.memory
         if memory is None:
-            memory = (
-                avail_memory
-                - object_store_memory
-                - (redis_max_memory if is_head else 0)
-            )
+            memory = avail_memory - object_store_memory
             if memory < 100e6 and memory < 0.05 * system_memory:
                 raise ValueError(
                     "After taking into account object store and redis memory "
@@ -311,7 +285,6 @@ def resolve(self, is_head: bool, node_ip_address: Optional[str] = None):
             memory,
             object_store_memory,
             resources,
-            redis_max_memory,
         )
         assert spec.resolved()
         return spec

python/ray/_private/worker.py

@@ -1422,7 +1422,6 @@ def init(
             created the object. Arguments to the task will be recursively
             reconstructed. If False, then ray.ObjectLostError will be
             thrown.
-        _redis_max_memory: Redis max memory.
         _plasma_directory: Override the plasma mmap file directory.
         _node_ip_address: The IP address of the node that we are on.
         _driver_object_store_memory: Deprecated.
@@ -1481,7 +1480,6 @@ def init(
     _enable_object_reconstruction: bool = kwargs.pop(
         "_enable_object_reconstruction", False
     )
-    _redis_max_memory: Optional[int] = kwargs.pop("_redis_max_memory", None)
     _plasma_directory: Optional[str] = kwargs.pop("_plasma_directory", None)
     _node_ip_address: str = kwargs.pop("_node_ip_address", None)
     _driver_object_store_memory: Optional[int] = kwargs.pop(
@@ -1726,7 +1724,6 @@ def sigterm_handler(signum, frame):
             dashboard_port=dashboard_port,
             memory=_memory,
             object_store_memory=object_store_memory,
-            redis_max_memory=_redis_max_memory,
             plasma_store_socket_name=None,
             temp_dir=_temp_dir,
             storage=storage,

python/ray/scripts/scripts.py

@@ -446,16 +446,6 @@ def debug(address: str, verbose: bool):
     "the shm size and 200G (ray_constants.DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES) "
     "but can be set higher.",
 )
-@click.option(
-    "--redis-max-memory",
-    required=False,
-    hidden=True,
-    type=int,
-    help="The max amount of memory (in bytes) to allow redis to use. Once the "
-    "limit is exceeded, redis will start LRU eviction of entries. This only "
-    "applies to the sharded redis tables (task, object, and profile tables). "
-    "By default this is capped at 10GB but can be set higher.",
-)
 @click.option(
     "--num-cpus", required=False, type=int, help="the number of CPUs on this node"
 )
@@ -668,7 +658,6 @@ def start(
     ray_client_server_port,
     memory,
     object_store_memory,
-    redis_max_memory,
     num_cpus,
     num_gpus,
     resources,
@@ -867,7 +856,6 @@ def start(
         # Initialize Redis settings.
         ray_params.update_if_absent(
             redis_shard_ports=redis_shard_ports,
-            redis_max_memory=redis_max_memory,
             num_redis_shards=num_redis_shards,
             redis_max_clients=None,
         )