[core][state] Task events backend - worker task event buffer implementation [1/n]

[rickyyx]

Why are these changes needed?

For details of the design and background see this doc

Previous PRs:

• [core][state] Task events backend: config and interface definitions [0/n]  #30829: Interface and protobuf definitions.

In this PR:

• Added a TaskEventBuffer class which serves as an abstraction to store task events, and push those events to GCS in batches periodically.
• Each CoreWorker will own one single TaskEventBuffer, and events (both task status change events and profiling events) will be added to a local in-memory buffer.
• The TaskEventBuffer also owns its own GCS client and io thread, which is independent from the main io_contexts.

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

[rickyyx]

Local benchmarking:

Some profiling results:

This is a result of running

n = 10000
m = 4
actors = [Actor.remote() for _ in range(m)]


def multi_task():
    submitted = [a.small_value_batch.remote(n) for a in actors]
    ray.get(submitted)

• The overhead mostly comes from memory related ops (constructing and destructing) of grpc structs when recording status changes in the main io thread.

[ericl]

Looks great--- main comment is around unifying some of the code for SetStatus.

[ericl]

For the performance change, I wonder if we can make the following unrelated optimization and gain back most of the difference: #30872

Basically, serialize the scheduling strategy protos to binary and compare those instead of using the message differencer.