Repository navigation
[Data] Terminal-only waits are counted as task output backpressure #67312
Description
Problem / Description
While reviewing #67180, I noticed a separate, pre-existing issue in DataOpTask's output-backpressure accounting.
DataOpTask._track_task_output_backpressure() currently starts the output-backpressure timer whenever max_bytes_to_read == 0.
However, a zero read budget does not necessarily mean that the task has unread output.
For example, an empty streaming task can already have its stream fully exhausted, with no block or metadata output left to consume, while the driver is only waiting to observe terminal completion.
If on_data_ready() is called with a zero output-read budget during this period, the task still accumulates task_output_backpressure_s.
This means time spent waiting only for terminal confirmation can be reported as output backpressure even though no output is waiting to be read.
This issue is separate from #67180 and reproduces on the unmodified current implementation.
Minimal Reproduction
A minimal internal-API regression case is:
streaming_gen = create_stub_streaming_gen(block_nbytes=[])
task = DataOpTask(
0,
streaming_gen,
BlockRefCounter(),
"test_op",
)
ray.get(streaming_gen.completed(), timeout=20)
assert streaming_gen._stream_exhausted()
clock = [100.0]
monkeypatch.setattr(
"ray.data._internal.execution.interfaces.physical_operator.time.perf_counter",
lambda: clock[0],
)
fetcher = InlineMetadataFetcher()
# No output exists; only terminal completion remains.
assert task.on_data_ready(0, fetcher) == 0
clock[0] += 10
# Allow the current implementation to observe EOF and finish.
assert task.on_data_ready(1, fetcher) == 0
assert task.has_finished
assert task._total_output_backpressure_s == 0
Actual behavior (unmodified Ray):
task._total_output_backpressure_s == 10.0
The task reports 10 seconds of output backpressure even though the streaming generator produced no output.
Expected behavior:
task._total_output_backpressure_s == 0
Waiting only for terminal completion should not count as output backpressure when there is no unread block or metadata output.
Actual unread output should continue to count normally, including the case where the generator stream is already exhausted but a block/metadata pair has already been pulled and remains pending inside DataOpTask.
Proposal
Track output backpressure only when both conditions are true:
- max_bytes_to_read == 0
- the task still has unread output.
A task still has unread output when at least one of the following holds:
- _pending_block_ref is non-ni;
- _pending_meta_ref is non-nil
- the streaming generator has not yet been exhausted.
If a backpressure interval is already open and these conditions stop being true, close the interval even if the output-read budget remains zero.
This prevents terminal-only waiting time from being included in the metric while preserving accounting for actual pending output.
Metadata
Metadata
Assignees
Labels
Type
Projects
- StatusShow more project fieldsDone
Activity
Jadiouo commentedon Oct 24, 2026
@Yin-1022 Are you planning to work on this? If not, I'd be happy to take it.
Yin-1022 commentedon Oct 24, 2026
Hi, thanks but I've almost finished and been ready to submit a pr.
I can tag you to help review afterward if you want to!
Jadiouo commentedon Oct 24, 2026
Sounds good, please tag me when the PR is up and I'll review it.
Yin-1022 commentedon Oct 24, 2026
The pr is up here! Thanks a lot
#67374