[core][state] Adjust worker side reporting with batches && add debugstring

[rickyyx]

Signed-off-by: rickyyx rickyx@anyscale.com

Why are these changes needed?

1. This PR introduces a flag RAY_task_events_send_batch_size that controls the number of task events sent to GCS in a batch. With default setting, each core worker will send 10K task events per second to GCS, where GCS could handle 10K task events in ~50 milliseconds.

2. This PR also adjust the worker side buffer limit to 1M with the new batching setting.

3. The PR adds some debug informations as well.

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

[rkooo567]

What about just sorting and popping? The current impl seems to have a couple tricky issues (read comments below) E.g.,

i = 0

while (i < next_index) {
    // Append  the first entry to the end until the next index
    buffer_.append(buffer_.pop(0))
    i += 1
}
next_index = 0

reported = 0
to_send = []
while (reported < batch_size) {
    to_send.append(buffer_.pop(0))
}

[rickyyx]

• Added test to assert sent order
• Used circular buffer as the buffer for sending and overwriting task events

[rkooo567]

One last question!

[rkooo567]

Awesome! Really happy with the final outcome!