Repository: ray-project/ray
Pull request: [core][compiled-graphs] Read input channels in a round-robin manner with a short timeout, and fail fast if a RayTaskError is found
<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

This PR is an alternative solution for #49444. See the comment https://github.com/ray-project/ray/pull/49444#issuecomment-2576266044 for more context.

This PR reads each channel in a round-robin manner with a shorter timeout (100 ms) and returns immediately if a RayTaskError is found.

* The read operation should be idempotent and have no side effects if it times out.
* NCCL send/recv operations do not support timeouts, so if an NCCL operation depends on other input channels, this PR cannot handle it. However, the end-to-end timeout will cover such cases.

In short, this PR has a smaller scope compared to #49444, but the solution is much easier and more lightweight.

* The behavior of `timeout` in `_read_list` has changed. The timeout value will now always be enforced. Previously, if the total time exceeded `timeout`, `_read_list` would still call the channel’s `read` function with `timeout=0`, attempting to acquire the object one final time. With this PR, once the total time exceeds the `timeout` specified in `_read_list`, a channel timeout exception is raised—regardless of whether the channel read operation has timed out.

* Wrap the `RayTaskError` in a `RayChannelFastFailError` that includes the channel index. When users call `ray.get` on one of the output references, an error log is printed, and the original exception (e.g., `ValueError` in the tests) is thrown instead of `RayChannelFastFailError` or `RayTaskError`.

## Related issue number

Closes #46337

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
    - [ ] I've added any new APIs to the API Reference. For example, if I added a 
           method in Tune, I've added it in `doc/source/tune/api/` under the 
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(


Problem statement:
[core][experimental] ray.get of accelerated DAG result may not throw exception for MultiOutputNode
### What happened + What you expected to happen

When calling `ray.get()` on the outputs of an accelerated DAG with a MultiOutputNode, we currently block on the results one at a time. This means that if one of the actor tasks threw an exception, we do not throw it until after we get the results for all of the previous actors' results, according to the order in the MultiOutputNode. This is a problem for tensor-parallel inference, because all of the workers execute in lockstep, and if one actor throws an exception, the others may hang. Depending on the order of the actors, `ray.get()` may never return.

We should fix this by trying to simultaneously get all of the outputs in the MultiOutputNode. One way is to try to `ReadAcquire` the channels in round-robin order.

### Versions / Dependencies

3.0dev

### Reproduction script

--

### Issue Severity

None

Hints from issue discussion:
Marking as p0 because it's important for vLLM use case. But it's okay to release developer preview without it.
@ruisearch42  are you tracking this?
Thanks @anyscalesam . Yes I will work on this.
I have a reproduction script here: https://gist.github.com/kevin85421/a7f14ea38d64420b105fbd79fd31fb8a

```python
with InputNode() as inp:
    # fail: raise an exception, sleep: sleep $inp seconds
    dag = MultiOutputNode([b.fail.bind(inp), a.sleep.bind(inp)]) 
compiled_dag = dag.experimental_compile()
ref = compiled_dag.execute(10)
ray.get(ref, timeout=30)
```

There are two input channels for `_dag_output_fetcher`: one for `b.fail` and one for `a.sleep`.

## Current behavior

Although `b.fail` throws an exception, the read operation still needs to wait until `a.sleep` finishes (in this example, 10 seconds) or until a timeout occurs (in this example, 30 seconds).

## Use two threads to read these two input channels (TL;DR: not recommended)

My first thought is to use two threads to read from these two input channels in `_read_list`, return immediately if either thread encounters a `RayTaskError`, and then stop any other threads. I implemented something like below:

```python
def _read_list(self, timeout: Optional[float] = None) -> List[Any]:
    results = [None] * len(self._input_channels)

    def _read_channel(idx: int, channel: ChannelInterface, timeout: Optional[float]):
        results[idx] = channel.read(timeout)
        return results[idx]

    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = []
        for idx, channel in enumerate(self._input_channels):
            futures.append(executor.submit(_read_channel, idx, channel, timeout))

        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            # if result is `RayTaskError`, cancel any other threads and return immediately.
        return results
```

However, I found that [concurrent.futures.cancel](https://docs.python.org/3/library/concurrent.futures.html) can't cancel running threads. Therefore, we still need to wait 10 seconds for `a.sleep` to finish before the function can return.

We currently have two options if we use multiple threads:
* Use `threading.Event()` to control when other threads stop reading. However, this requires many changes, from Python and Cython to C++.
* Use `future.result(timeout=...)`. However, we have already supported timeout.

In addition, in my benchmark, multi-threading causes several times performance degradation in `_read_list` if the input size is not large enough, and only shows small performance improvements with larger input sizes.

* TL;DR: not recommended

# Recommended solution

1. Use the original `_read_list` and check whether the result is a `RayTaskError` after each read operation. This way, some cases will return earlier if `fail` is read before `sleep`.

2. For cases not covered by (1), wait for a timeout.

The drawback is that users need to specify timeout by themselves.
> Use the original _read_list and check whether the result is a RayTaskError after each read operation. This way, some cases will return earlier if fail is read before sleep.

Does it work when you do

```
dag = MultiOutputNode([a.sleep.bind(inp), b.fail.bind(inp)]) 
```
?

I think what we need is try batch waiting with short timeout (and keep checking until every objects are ready). similar to how ray.get is implemented in ray. (See `Status CoreWorkerMemoryStore::GetImpl`)
> Does it work when you do `dag = MultiOutputNode([a.sleep.bind(inp), b.fail.bind(inp)])`?

No

> I think what we need is try batch waiting with short timeout (and keep checking until every objects are ready). similar to how ray.get is implemented in ray. (See `Status CoreWorkerMemoryStore::GetImpl`)

Are you referring to https://sourcegraph.com/github.com/ray-project/ray@6309e4be65fe94ed9489f3b035a6ba1215e71095/-/blob/src/ray/core_worker/store_provider/memory_store/memory_store.cc?L364-L375?

Do you mean:
1. Using multiple threads to read each object in `MultiOutputNode`.
2. Specifying a shorter timeout for `channel.read`.
3. If `channel.read` times out, check whether it exceeds the timeout specified by the user. If not, retry.

Commit message: address comments

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

Patch to review:
diff --git a/python/ray/dag/context.py b/python/ray/dag/context.py
index 25fbd8bd173b0..1b8f23f5dfa65 100644
--- a/python/ray/dag/context.py
+++ b/python/ray/dag/context.py
@@ -11,6 +11,9 @@
 DEFAULT_SUBMIT_TIMEOUT_S = int(os.environ.get("RAY_CGRAPH_submit_timeout", 10))
 DEFAULT_GET_TIMEOUT_S = int(os.environ.get("RAY_CGRAPH_get_timeout", 10))
 DEFAULT_TEARDOWN_TIMEOUT_S = int(os.environ.get("RAY_CGRAPH_teardown_timeout", 30))
+DEFAULT_READ_ITERATION_TIMEOUT_S = float(
+    os.environ.get("RAY_CGRAPH_read_iteration_timeout_s", 0.1)
+)
 # Default buffer size is 1MB.
 DEFAULT_BUFFER_SIZE_BYTES = int(os.environ.get("RAY_CGRAPH_buffer_size_bytes", 1e6))
 # Default asyncio_max_queue_size is 0, which means no limit.
@@ -57,6 +60,10 @@ class DAGContext:
             value higher than the expected time to execute the entire DAG.
         teardown_timeout: The maximum time in seconds to wait for the DAG to
             cleanly shut down.
+        read_iteration_timeout: The timeout in seconds for each read iteration
+            that reads one of the input channels. If the timeout is reached, the
+            read operation will be interrupted and will try to read the next
+            input channel. It must be less than or equal to `get_timeout`.
         buffer_size_bytes: The initial buffer size in bytes for messages
             that can be passed between tasks in the DAG. The buffers will
             be automatically resized if larger messages are written to the
@@ -83,12 +90,21 @@ class DAGContext:
     submit_timeout: int = DEFAULT_SUBMIT_TIMEOUT_S
     get_timeout: int = DEFAULT_GET_TIMEOUT_S
     teardown_timeout: int = DEFAULT_TEARDOWN_TIMEOUT_S
+    read_iteration_timeout: float = DEFAULT_READ_ITERATION_TIMEOUT_S
     buffer_size_bytes: int = DEFAULT_BUFFER_SIZE_BYTES
     asyncio_max_queue_size: int = DEFAULT_ASYNCIO_MAX_QUEUE_SIZE
     max_buffered_results: int = DEFAULT_MAX_BUFFERED_RESULTS
     max_inflight_executions: int = DEFAULT_MAX_INFLIGHT_EXECUTIONS
     overlap_gpu_communication: bool = DEFAULT_OVERLAP_GPU_COMMUNICATION
 
+    def __post_init__(self):
+        if self.read_iteration_timeout > self.get_timeout:
+            raise ValueError(
+                "RAY_CGRAPH_read_iteration_timeout_s "
+                f"({self.read_iteration_timeout}) must be less than or equal to "
+                f"RAY_CGRAPH_get_timeout ({self.get_timeout})"
+            )
+
     @staticmethod
     def get_current() -> "DAGContext":
         """Get or create a singleton context.
diff --git a/python/ray/dag/tests/experimental/test_accelerated_dag.py b/python/ray/dag/tests/experimental/test_accelerated_dag.py
index 00c476022b8a5..63fa4bb679e95 100644
--- a/python/ray/dag/tests/experimental/test_accelerated_dag.py
+++ b/python/ray/dag/tests/experimental/test_accelerated_dag.py
@@ -1849,6 +1849,228 @@ def test_intra_process_channel_with_multi_readers(
             assert ray.get(refs) == [3, 3]
 
 
+@ray.remote
+class FastFailActor:
+    def sleep_and_echo(self, x):
+        time.sleep(x)
+        return x
+
+    def fail_if_x_is_even(self, x):
+        if x % 2 == 0:
+            raise ValueError("x is even")
+        return x
+
+    def sleep_and_fail(self, x):
+        time.sleep(x)
+        raise ValueError("fail")
+
+
+class TestFastFail:
+    @pytest.mark.parametrize("is_async", [True, False])
+    def test_first_input_fail(self, ray_start_regular, is_async):
+        """
+        Tests the case where the failing input is at the beginning of the input list.
+        """
+        a = FastFailActor.remote()
+        with InputNode() as inp:
+            dag = MultiOutputNode(
+                [a.fail_if_x_is_even.bind(inp), a.sleep_and_echo.bind(inp)]
+            )
+        compiled_dag = dag.experimental_compile(enable_asyncio=is_async)
+
+        if is_async:
+
+            async def main():
+                futs = await compiled_dag.execute_async(6)
+                start_time = time.time()
+                with pytest.raises(ValueError, match="x is even"):
+                    await asyncio.gather(*futs)
+                end_time = time.time()
+                assert end_time - start_time < 6
+
+            loop = get_or_create_event_loop()
+            loop.run_until_complete(main())
+        else:
+            start_time = time.time()
+            with pytest.raises(ValueError, match="x is even"):
+                ray.get(compiled_dag.execute(6))
+            end_time = time.time()
+            assert end_time - start_time < 6
+
+    @pytest.mark.parametrize("is_async", [True, False])
+    def test_last_input_fail(self, ray_start_regular, is_async):
+        """
+        Tests the case where the failing input is at the end of the input list.
+        The test cannot use the same actor for both `sleep_and_echo` and
+        `fail_if_x_is_even` tasks because the control dependency would make the
+        `fail_if_x_is_even` task execute after the `sleep_and_echo` task finishes.
+        """
+        a = FastFailActor.remote()
+        b = FastFailActor.remote()
+        with InputNode() as inp:
+            dag = MultiOutputNode(
+                [a.sleep_and_echo.bind(inp), b.fail_if_x_is_even.bind(inp)]
+            )
+        compiled_dag = dag.experimental_compile(enable_asyncio=is_async)
+
+        if is_async:
+
+            async def main():
+                futs = await compiled_dag.execute_async(6)
+                start_time = time.time()
+                with pytest.raises(ValueError, match="x is even"):
+                    await asyncio.gather(*futs)
+                end_time = time.time()
+                assert end_time - start_time < 6
+
+            loop = get_or_create_event_loop()
+            loop.run_until_complete(main())
+        else:
+            start_time = time.time()
+            with pytest.raises(ValueError, match="x is even"):
+                ray.get(compiled_dag.execute(6))
+            end_time = time.time()
+            assert end_time - start_time < 6
+
+    @pytest.mark.parametrize("is_async", [True, False])
+    def test_get_one_of_output_refs(self, ray_start_regular, is_async):
+        """
+        Tests the case where `ray.get` is called on only one of the output refs
+        which doesn't fail.
+        """
+        a = FastFailActor.remote()
+        b = FastFailActor.remote()
+        with InputNode() as inp:
+            dag = MultiOutputNode(
+                [a.sleep_and_echo.bind(inp), b.fail_if_x_is_even.bind(inp)]
+            )
+        compiled_dag = dag.experimental_compile(enable_asyncio=is_async)
+
+        if is_async:
+
+            async def main():
+                futs = await compiled_dag.execute_async(6)
+                start_time = time.time()
+                with pytest.raises(ValueError, match="x is even"):
+                    await asyncio.gather(futs[0])
+                end_time = time.time()
+                assert end_time - start_time < 6
+
+            loop = get_or_create_event_loop()
+            loop.run_until_complete(main())
+        else:
+            start_time = time.time()
+            with pytest.raises(ValueError, match="x is even"):
+                refs = compiled_dag.execute(6)
+                ray.get(refs[0])
+            end_time = time.time()
+            assert end_time - start_time < 6
+
+    @pytest.mark.parametrize("is_async", [True, False])
+    def test_middle_input_fail(self, ray_start_regular, is_async):
+        """
+        Tests the case where the failing input is in the middle of the input list.
+        """
+        a = FastFailActor.remote()
+        b = FastFailActor.remote()
+        c = FastFailActor.remote()
+        with InputNode() as inp:
+            dag = MultiOutputNode(
+                [
+                    a.sleep_and_echo.bind(inp),
+                    b.fail_if_x_is_even.bind(inp),
+                    c.sleep_and_echo.bind(inp),
+                ]
+            )
+        compiled_dag = dag.experimental_compile(enable_asyncio=is_async)
+
+        if is_async:
+
+            async def main():
+                futs = await compiled_dag.execute_async(6)
+                start_time = time.time()
+                with pytest.raises(ValueError, match="x is even"):
+                    await asyncio.gather(*futs)
+                end_time = time.time()
+                assert end_time - start_time < 6
+
+            loop = get_or_create_event_loop()
+            loop.run_until_complete(main())
+        else:
+            start_time = time.time()
+            with pytest.raises(ValueError, match="x is even"):
+                ray.get(compiled_dag.execute(6))
+            end_time = time.time()
+            assert end_time - start_time < 6
+
+    @pytest.mark.parametrize("is_async", [True, False])
+    def test_all_inputs_fail(self, ray_start_regular, is_async):
+        """
+        Tests the case where all inputs fail with different sleep times.
+        """
+        a = FastFailActor.remote()
+        b = FastFailActor.remote()
+        c = FastFailActor.remote()
+        with InputNode() as inp:
+            dag = MultiOutputNode(
+                [
+                    a.sleep_and_fail.bind(inp[0]),
+                    b.sleep_and_fail.bind(inp[1]),
+                    c.sleep_and_fail.bind(inp[2]),
+                ]
+            )
+        compiled_dag = dag.experimental_compile(enable_asyncio=is_async)
+
+        if is_async:
+
+            async def main():
+                futs = await compiled_dag.execute_async(6, 0, 6)
+                start_time = time.time()
+                with pytest.raises(ValueError, match="fail"):
+                    await asyncio.gather(*futs)
+                end_time = time.time()
+                assert end_time - start_time < 6
+
+            loop = get_or_create_event_loop()
+            loop.run_until_complete(main())
+        else:
+            start_time = time.time()
+            with pytest.raises(ValueError, match="fail"):
+                ray.get(compiled_dag.execute(6, 0, 6))
+            end_time = time.time()
+            assert end_time - start_time < 6
+
+    @pytest.mark.parametrize("is_async", [True, False])
+    def test_fail_and_retry(self, ray_start_regular, is_async):
+        """
+        Tests the case where the first input fails, but the second input succeeds.
+        """
+        a = FastFailActor.remote()
+        with InputNode() as inp:
+            dag = MultiOutputNode(
+                [a.fail_if_x_is_even.bind(inp), a.sleep_and_echo.bind(inp)]
+            )
+        compiled_dag = dag.experimental_compile(enable_asyncio=is_async)
+
+        if is_async:
+
+            async def main():
+                futs = await compiled_dag.execute_async(2)
+                with pytest.raises(ValueError, match="x is even"):
+                    await asyncio.gather(*futs)
+                for _ in range(3):
+                    futs = await compiled_dag.execute_async(1)
+                    assert await asyncio.gather(*futs) == [1, 1]
+
+            loop = get_or_create_event_loop()
+            loop.run_until_complete(main())
+        else:
+            with pytest.raises(ValueError, match="x is even"):
+                ray.get(compiled_dag.execute(2))
+            for _ in range(3):
+                assert ray.get(compiled_dag.execute(1)) == [1, 1]
+
+
 class TestLeafNode:
     """
     Leaf nodes are not allowed right now because the exception thrown by the leaf
diff --git a/python/ray/experimental/channel/common.py b/python/ray/experimental/channel/common.py
index 65d4c6961cd3d..9261f59db9628 100644
--- a/python/ray/experimental/channel/common.py
+++ b/python/ray/experimental/channel/common.py
@@ -279,6 +279,12 @@ def __init__(
         self._closed = False
         self._num_reads = 0
 
+        # A list of channels that were not read in the last `read` call
+        # because the reader returned immediately when a RayTaskError was found.
+        # These channels must be consumed before the next read to avoid reading
+        # stale data remaining from the last read.
+        self._leftover_channels: List[ChannelInterface] = []
+
     def get_num_reads(self) -> int:
         return self._num_reads
 
@@ -291,8 +297,8 @@ def _read_list(self, timeout: Optional[float] = None) -> List[Any]:
 
         Args:
             timeout: The maximum time in seconds to wait for reading.
-                None means using default timeout, 0 means immediate timeout
-                (immediate success or timeout without blocking), -1 means
+                None means using default timeout which is infinite, 0 means immediate
+                timeout (immediate success or timeout without blocking), -1 means
                 infinite timeout (block indefinitely).
 
         """
@@ -320,6 +326,22 @@ def close(self) -> None:
         for channel in self._input_channels:
             channel.close()
 
+    def _consume_leftover_channels_if_needed(
+        self, timeout: Optional[float] = None
+    ) -> None:
+        # Consume the channels that were not read in the last `read` call because
+        # a RayTaskError was found. If we don't do this, the read operation will read
+        # stale versions of the object refs.
+        #
+        # TODO(kevin85421): Currently, a DAG with NCCL channels and fast fail enabled
+        # may not be reusable. Revisit this in the future.
+        for c in self._leftover_channels:
+            start_time = time.monotonic()
+            c.read(timeout)
+            if timeout is not None:
+                timeout -= time.monotonic() - start_time
+                timeout = max(timeout, 0)
+
 
 @DeveloperAPI
 class SynchronousReader(ReaderInterface):
@@ -333,13 +355,63 @@ def start(self):
         pass
 
     def _read_list(self, timeout: Optional[float] = None) -> List[Any]:
-        results = []
-        for c in self._input_channels:
-            start_time = time.monotonic()
-            results.append(c.read(timeout))
-            if timeout is not None:
-                timeout -= time.monotonic() - start_time
-                timeout = max(timeout, 0)
+        self._consume_leftover_channels_if_needed(timeout)
+        # We don't update `remaining_timeout` here because in the worst case,
+        # consuming leftover channels requires reading all `_input_channels`,
+        # which users expect to complete within the original `timeout`. Updating
+        # `remaining_timeout` could cause unexpected timeouts in subsequent read
+        # operations.
+        self._leftover_channels = []
+
+        # It is a special case that `timeout` is set to 0, which means
+        # read once for each channel.
+        is_zero_timeout = timeout == 0
+
+        results = [None for _ in range(len(self._input_channels))]
+        if timeout is None or timeout == -1:
+            timeout = float("inf")
+        timeout_point = time.monotonic() + timeout
+        remaining_timeout = timeout
+
+        from ray.dag import DAGContext
+
+        ctx = DAGContext.get_current()
+        iteration_timeout = ctx.read_iteration_timeout
+
+        # Iterate over the input channels with a shorter timeout for each iteration
+        # to detect RayTaskError early and fail fast.
+        done_channels = set()
+        while len(done_channels) < len(self._input_channels):
+            for i, c in enumerate(self._input_channels):
+                if c in done_channels:
+                    continue
+                try:
+                    result = c.read(min(remaining_timeout, iteration_timeout))
+                    results[i] = result
+                    done_channels.add(c)
+                    if isinstance(result, ray.exceptions.RayTaskError):
+                        # If we raise an exception immediately, it will be considered
+                        # as a system error which will cause the execution loop to
+                        # exit. Hence, return immediately and let `_process_return_vals`
+                        # handle the exception.
+                        #
+                        # Return a list of RayTaskError so that the caller will not
+                        # get an undefined partial result.
+                        self._leftover_channels = [
+                            c for c in self._input_channels if c not in done_channels
+                        ]
+                        return [result for _ in range(len(self._input_channels))]
+                except ray.exceptions.RayChannelTimeoutError as e:
+                    remaining_timeout = max(timeout_point - time.monotonic(), 0)
+                    if remaining_timeout == 0:
+                        raise e
+                    continue
+
+                remaining_timeout = max(timeout_point - time.monotonic(), 0)
+                if remaining_timeout == 0 and not is_zero_timeout:
+                    raise ray.exceptions.RayChannelTimeoutError(
+                        f"Cannot read all channels within {timeout} seconds"
+                    )
         return results
 
     def release_channel_buffers(self, timeout: Optional[float] = None) -> None:
@@ -377,14 +449,36 @@ def start(self):
         self._background_task = asyncio.ensure_future(self.run())
 
     def _run(self):
-        results = []
-        for c in self._input_channels:
-            exiting = retry_and_check_interpreter_exit(
-                lambda: results.append(c.read(timeout=1))
-            )
-            if exiting:
-                break
-
+        # Give it a default timeout 60 seconds to release the buffers
+        # of the channels that were not read in the last `read` call.
+        self._consume_leftover_channels_if_needed(60)
+        self._leftover_channels = []
+
+        results = [None for _ in range(len(self._input_channels))]
+
+        from ray.dag import DAGContext
+
+        ctx = DAGContext.get_current()
+        iteration_timeout = ctx.read_iteration_timeout
+
+        done_channels = set()
+        while len(done_channels) < len(self._input_channels):
+            for i, c in enumerate(self._input_channels):
+                if c in done_channels:
+                    continue
+                try:
+                    result = c.read(iteration_timeout)
+                    results[i] = result
+                    done_channels.add(c)
+                    if isinstance(result, ray.exceptions.RayTaskError):
+                        self._leftover_channels = [
+                            c for c in self._input_channels if c not in done_channels
+                        ]
+                        return [result for _ in range(len(self._input_channels))]
+                except ray.exceptions.RayChannelTimeoutError:
+                    pass
+                if sys.is_finalizing():
+                    return results
         return results
 
     async def run(self):
@@ -450,10 +544,9 @@ def write(self, val: Any, timeout: Optional[float] = None) -> None:
         Write the value.
 
         Args:
-            timeout: The maximum time in seconds to wait for writing.
-                None means using default timeout, 0 means immediate timeout
-                (immediate success or timeout without blocking), -1 means
-                infinite timeout (block indefinitely).
+            timeout: The maximum time in seconds to wait for writing. 0 means
+                immediate timeout (immediate success or timeout without blocking).
+                -1 and None mean infinite timeout (blocks indefinitely).
         """
         raise NotImplementedError()
 


Code review comments:
diff_hunk: @@ -333,13 +355,63 @@ def start(self):
         pass
 
     def _read_list(self, timeout: Optional[float] = None) -> List[Any]:
-        results = []
-        for c in self._input_channels:
-            start_time = time.monotonic()
-            results.append(c.read(timeout))
-            if timeout is not None:
-                timeout -= time.monotonic() - start_time
-                timeout = max(timeout, 0)
+        self._consume_leftover_channels_if_needed(timeout)
+        # We don't update `remaining_timeout` here because in the worst case,
+        # consuming leftover channels requires reading all `_input_channels`,
+        # which users expect to complete within the original `timeout`. Updating
+        # `remaining_timeout` could cause unexpected timeouts in subsequent read
+        # operations.
+        self._leftover_channels = []
path: python/ray/experimental/channel/common.py
text: @user1:
fold this into `_consume_leftover_channels_if_needed`

@author:
updated: https://github.com/ray-project/ray/pull/49711/commits/efeb583d5160876dd41a3ca4b7b3018ad65af762