[data] add backpressure reason

[Jay-ju]

Background

The concurrency of raydata's operators is mostly set manually, and the automatic perception mostly cannot meet the situation of resource utilization. However, manually setting the operator concurrency is very easy to trigger rate limiting, but it is currently difficult to obtain the reason for rate limiting.

Backpressure trigger conditions

There are a total of four trigger conditions:

• Reserved resources: DataContext.get_current()data_context.op_resource_reservation_enabled = true (default). Each time a new task is submitted, it will check whether the resources required by the task can still be met. The overall resources will be divided into reserved and shared.
• Unreserved resources: DataContext.get_current()data_context.op_resource_reservation_enabled = false. Resources are managed uniformly.
• Backpressure policy: In the open-source policy, only ConcurrencyCapBackpressurePolicy is supported, whether the number of running tasks exceeds the set concurrency number.
• Whether the Op can increase input: For the actor pool, whether there is remaining free slots space, mainly depending on the difference between the number of running tasks on the actor and the default number.

Backpressure enhancement

Backpressure observability enhancement

• Before enhancement:
  

• After enhancement

  • Concurrent backpressure policy
    

  • Backpressure in non-reserved mode
    

  • Backpressure in reserved mode
    

  • Insufficient free slots of actor
    

Why are these changes needed?

Related issue number

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 :(

[raulchen]

Can you add PR description? Ideally also add some screenshots. Thanks.
My main concern is that this can add too much content to the progress bar, making it truncated.
cc @scottjlee @omatthew98 who are actively improving progress bars for review.

[scottjlee]

The reason string content generally makes sense to me. My main concern is that the progress bar outputs will become very verbose, making it difficult to read. Also, the notation with G, D, without explanation seems it would be confusing to users.

To address both of these points, I think we should configure this as an advanced feature, and disable by default. The user should enable seeing the backpressure reason through DataContext, e.g. DataContext.get_current().show_backpressure_reason = True. We should also add docs to explain how to interpret the reasoning output -- you can add this to the Ray Data progress bar section.

(Ray Data doesn't explicitly truncate the stats outputs for each operator, only truncates the operator name if it is too long).

[Jay-ju]

The reason string content generally makes sense to me. My main concern is that the progress bar outputs will become very verbose, making it difficult to read. Also, the notation with G, D, without explanation seems it would be confusing to users.

To address both of these points, I think we should configure this as an advanced feature, and disable by default. The user should enable seeing the backpressure reason through DataContext, e.g. DataContext.get_current().show_backpressure_reason = True. We should also add docs to explain how to interpret the reasoning output -- you can add this to the Ray Data progress bar section.

(Ray Data doesn't explicitly truncate the stats outputs for each operator, only truncates the operator name if it is too long).

Thank you for your review. I have made the changes according to your suggestions.

[alexeykudinkin]

Let's wrap this into new structure like:

self._tasks_state = TaskState(...)

class TaskState:
  submission_throttled: bool
  submission_throttled_reason: str = ...

  # ...

[Jay-ju]

done