Add runtime env port range conf (#682)

* Add runtime env agent port range conf.

* fix

* Apply suggestion from @gemini-code-assist[bot]

fix get env

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: liying919

python/ray/_private/node.py

@@ -988,12 +988,19 @@ def _get_unused_port(self, allocated_ports=None):
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         s.bind(("", 0))
         port = s.getsockname()[1]
+        low_end = ray_constants.env_integer("RAY_PORT_RANGE_LOW", port)
+        high_end = ray_constants.env_integer("RAY_PORT_RANGE_HIGH", 65535)
+        if low_end > high_end:
+            raise ValueError(
+                f"Invalid port range: RAY_PORT_RANGE_LOW ({low_end}) must be less than or equal to RAY_PORT_RANGE_HIGH ({high_end})."
+            )
 
-        # Try to generate a port that is far above the 'next available' one.
-        # This solves issue #8254 where GRPC fails because the port assigned
-        # from this method has been used by a different process.
+        # Try to generate a port that is far above the 'next available' one
+        # or from a given range(if low_end and high_end is defined). This
+        # solves issue #8254 where GRPC fails because the port assigned from
+        # this method has been used by a different process.
         for _ in range(ray_constants.NUM_PORT_RETRIES):
-            new_port = random.randint(port, 65535)
+            new_port = random.randint(low_end, high_end)
             if new_port in allocated_ports:
                 # This port is allocated for other usage already,
                 # so we shouldn't use it even if it's not in use right now.