[Data] Concurrency Cap Backpressure tuning

[srinathk10]

Thank you for contributing to Ray! 🚀
Please review the Ray Contribution Guide before opening a pull request.

⚠️ Remove these instructions before submitting your PR.

💡 Tip: Mark as draft if you want early feedback, or ready for review when it's complete.

Description

Briefly describe what this PR accomplishes and why it's needed.

[Data] Concurrency Cap Backpressure tuning

• Maintain asymmetric EWMA of total queued bytes (this op + downstream) as the typical level: level.
• Maintain asymmetric EWMA of absolute residual vs the previous level as a scale proxy: dev = EWMA(|q - level_prev|).
• Define deadband: [lower, upper] = [level - K_DEVdev, level + K_DEVdev].
  If q > upper -> target cap = running - BACKOFF_FACTOR (back off)
  If q < lower -> target cap = running + RAMPUP_FACTOR (ramp up)
  Else -> target cap = running (hold)
• Clamp to [1, configured_cap], admit iff running < target cap.

Related issues

Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234".

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[srinathk10]

Benchmark

Baseline vs After

+---------------------------------------------+--------+----------+
| Benchmark                                   | After  | Baseline |
+---------------------------------------------+--------+----------+
| skip_training.parquet                       | 11257  | 10924    |
| skip_training.parquet.preserve_order        | 10754  | 10611    |
| skip_training.jpeg                          | 3325   | 3445     |
| skip_training.jpeg.preserve_order           | 1584   | 1775     |
| skip_training.jpeg.local_fs                 | 2042   | 2027     |
| skip_training.jpeg.local_fs.preserve_order  | 1987   | 1975     |
| skip_training.jpeg.local_fs_multi_gpus      | 2623   | 2658     |
| skip_training.jpeg.local_fs_multi_gpus.preserve_order | 2592 | 2615 |
+---------------------------------------------+--------+----------+

[alexeykudinkin]

Hold on, what's this for?

[srinathk10]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors the ConcurrencyCapBackpressurePolicy to use a simpler deadband-based algorithm for adjusting concurrency, which improves clarity and maintainability. The changes also make the policy's parameters configurable via environment variables.

My review focuses on the correctness of the new algorithm and its implementation. I've found a high-severity issue where the object store memory pressure check appears to be logically inverted, which could lead to incorrect backpressure behavior. I've also pointed out a related confusing comment and a minor typo in the docstring. Additionally, the corresponding unit test for the memory pressure check will need to be updated once the main logic is fixed.

Overall, this is a positive change that simplifies the system, and with the suggested fixes, it will be a solid improvement.

[raulchen]

Approving as this change per se makes sense.
But I think we need to revisit the whole policy again, after we fix all the accounting issues, e.g., prefetched data.
Because when that is done, this should no longer be needed to prevent spilling.
But this may still be useful for smoothing out sudden jumps.
So I'd suggest keeping it experimental for now and do more experiments.
Also, we should separate out the smoothing logic as standalone backpressure policy.