[core] Fix the race condition in the new resource broadcasting.

[fishbone]

Why are these changes needed?

The root cause is that disconnect is called not in io_context which in the end accessing the invalid memory. Disconnect needs to be called exactly once.

In gRPC, read/write might error and OnCancel also means error just with different semantics. But we need to ensure disconnect is only called once. Previously, disconnect might be called in gRPC threads, main thread/... and this PR enforce them called within the io context.

This fix the release test long_running_node_failures.

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

[fishbone]

It's a race condition and so hard to add a unit test for this. let me know if anyone has good ideas.

[fishbone]

release testing with fixing:
https://buildkite.com/ray-project/release-tests-pr/builds/29146#01868037-45c1-4237-bc65-d182baf991e3

[rkooo567]

The change itself lgtm. But this makes me think it's probably better making the module thread-safe rather than relying on posting to the io service? Seems like the threading model is very prone to error rn (I know it is a common problem in Ray, but I am bringing this up because it is a new module and we can enforce some good practice here?)

[fishbone]

@rkooo567 making it thread safe is the same as posting things to io-context. Basically, you can replace posting => mutex lock. So that's why I'm not putting mutex there.

And issues come from gRPC's callback is in gRPC thread. A good practice from gRPC is that don't hold the thread for long time.

IMPORTANT USAGE NOTE : code in any reaction must not block for an arbitrary amount of time since reactions are executed on a finite-sized, library-controlled threadpool. If any long-term blocking operations (like sleeps, file I/O, synchronous RPCs, or waiting on a condition variable) must be invoked as part of the application logic, then it is important to push that outside the reaction so that the reaction can complete in a timely fashion. One way of doing this is to push that code to a separate application-controlled thread.

The concerns or the difficulties should be handled in gRPC/threading related work.

[fishbone]

Hmmm, the test entered the terminated status... Seems the cluster is terminated... I'll check what's going on there.

[fishbone]

I think it might be related to the migration to jobs on v2. Let me check.

[fishbone]

Confirmed that it's an infra issue. Merge this one.