[core] [1/N] Ray syncer set observer

[dentiny]

Resolves issue: #48837

This PR is the first part for reducing communication between GCS/raylet.
The functionality is:

• Provide an observer callable to ray syncer and node state
• Every time bidirectional grpc is completed, invoke the registered callback
• As for the callback, I plan to pass in functor to update <node id, updated timestamp> which is stored in health check manager (in next PR)

[dentiny]

Hi @jjyao , could you please help review this PR? Thank you!

[jjyao]

Can we instead define a generic

// A callback is called whenver a rpc completes between the current process and the given raylet regardless of whether the response indicates success or failure
using RayletCompletedRpcCallback(void(const NodeID &))

and this callback can be set on any components that have rpcs with raylet.

[dentiny]

regardless of whether the response indicates success or failure

I'm not sure this is correct. For example, if you get an error status on grpc (i.e. UNAVAILABLE for grpc status), it actually means communication between gcs and raylet is broken, which we shouldn't record completion.

[jjyao]

Here Completed means client sends the request and then receives the response so if this is UNAVAILABLE, it's not completed. This is the naming suggested by chatgpt lol.

[jjyao]

Or we can think Completed == GrpcStatus::OK

[jjyao]

But feel free to give a better name.

[dentiny]

Updated.

[dentiny]

I removed Raylet from the comment and type alias, the whole RaySyncer is pretty general-purpose, which doesn't couple with any component in terms of implementation.

[jjyao]

I think the best place to trigger the callback is RaySyncerBidiReactorBase::OnReadDone() because NodeState::ConsumeSyncMessage can be called in different settings.

[dentiny]

Sounds good, I think it's good to update right after a successful grpc call.

[dentiny]

Ok I recall why I put it in the NodeState at the first place, reactors are stored per-node as a map inside of RaySyncer:

ray/src/ray/common/ray_syncer/ray_syncer.h

Lines 162 to 166 in 2e4a126

	/// Manage connections. Here the key is the NodeID in binary form.
	absl::flat_hash_map<std::string, RaySyncerBidiReactor *> sync_reactors_;
	/// The local node state
	std::unique_ptr<NodeState> node_state_;

which means if you tie the logic grpc callback, you have to store it somewhere and pass it to every reactor construction.

[dentiny]

Implementation updated.

[jjyao]

LG

[jjyao]

Suggested change

	/// This function is expected to call only once, repeated invocations throws exception.
	/// This function is expected to call only once, repeated invocations will check fail.

[dentiny]

I'm not sure why "will check fail" is a better wording?

[dentiny]

Anyway I updated the wording, but would like to hear the reasoning.

[jjyao]

check fail will just exit the process so no c++ exception is thrown?