[scheduling][5] Refactor resource syncer. #23270

fishbone · 2022-03-17T05:28:30Z

Why are these changes needed?

This PR refactor the resource syncer to decouple it from GCS and raylet. GCS and raylet will use the same module to sync data. The integration will happen in the next PR.

There are several new introduced components:

RaySyncer: the place where remote and local information sits. It's a coordinator layer.
NodeState: keeps track of the local status, similar to NodeSyncConnection.
NodeSyncConnection: keeps track of the sending and receiving information and make sure not sending the information the remote node knows.

The core protocol is that each node will send {what it has} - {what the target has} to the target.
For example, think about node A <-> B. A will send all A has exclude what B has to B.

Whenever when there is new information (from NodeState or NodeSyncConnection), it'll be passed to RaySyncer broadcast message to broadcast.

NodeSyncConnection is for the communication layer. It has two implementations Client and Server:

Server => Client: client will send a long-polling request and server will response every 100ms if there is data to be sent.
Client => Server: client will check every 100ms to see whether there is new data to be sent. If there is, just use RPC call to send the data.

Here is one example:

flowchart LR;
    A-->B;
    B-->C;
    B-->D;

It means A initialize the connection to B and B initialize the connections to C and D

Now C generate a message M:

[C] RaySyncer check whether there is new message generated in C and get M
[C] RaySyncer will push M to NodeSyncConnection in local component (B)
[C] ServerSyncConnection will wait until B send a long polling and send the data to B
[B] B received the message from C and push it to local sync connection (C, A, D)
[B] ClientSyncConnection of C will not push it to its local queue since it's received by this channel.
[B] ClientSyncConnection of D will send this message to D
[B] ServerSyncConnection of A will be used to send this message to A (long-polling here)
[B] B will update NodeState (local component) with this message M
[D] D's pipelines is similar to 5) (with ServerSyncConnection) and 8)
[A] A's pipeline is similar to 5) and 8)

Related issue number

Checks

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 · 2022-03-17T07:42:59Z

Still working on writing the test cases and toy example. Given that this PR will be too big, the actual integration will happen in the next PR.

src/ray/protobuf/ray_syncer.proto

src/ray/common/ray_syncer.h

…r-syncer

src/ray/common/ray_syncer.h

fishbone

Synced offline with @scv119 and the following concerns should be addressed in the updates:

We should put local node status into a separate class and provide Updates to update local status. (cluster_view_/reporters_/receivers_)
Consider using upstream/downstream model.

fishbone · 2022-03-18T08:54:22Z

Several improvements to address the comments.

Local node info is put into a separate structure
Hide all roll-related behavior to the communication layer
Move internal API to another header file to make the public API clean.

The next step:

Re-iterate the API with @scv119
Make toy example working
More unit tests
Make the PR ready for formal review and collect feedback from more people
(following PR) integrate with gcs and raylet.

scv119 · 2022-03-18T17:33:09Z

src/ray/protobuf/ray_syncer.proto

+  // The component this message is for.
+  RayComponentId component_id = 2;
+  // The actual payload.
+  bytes sync_message = 3;


will this incur one more string copy? we probably can benchmark it. if it's a concern we can see if union makes it better.

We have one extra copy to avoid one extra serialization when sending data.

Furthermore, I think in the future, the component actually can use flatbuffer to optimize for performance if needed.

jjyao

Didn't finish, will continue.

src/ray/common/ray_syncer/ray_syncer-inl.h

jjyao · 2022-03-25T22:24:14Z

src/ray/common/ray_syncer/ray_syncer-inl.h

+
+  RaySyncer &instance_;
+  instrumented_io_context &io_context_;
+  std::string node_id_;


this is local_node_id or target_node_id?

this is the node id of the connection. do you think the target node makes sense here?

jjyao · 2022-03-25T22:28:19Z

src/ray/common/ray_syncer/ray_syncer-inl.h

+  /// It'll send nothing unless there is a request from the remote node
+  /// for the sending request.


Got it. Maybe make the comment a bit clearer: like "send nothing unless there is a long-polling request ..."?

src/ray/common/ray_syncer/ray_syncer-inl.h

src/ray/common/ray_syncer/ray_syncer.cc

jjyao · 2022-03-25T23:24:47Z

src/ray/common/ray_syncer/ray_syncer-inl.h

+
+  absl::flat_hash_set<std::shared_ptr<const RaySyncMessage>, _MessageHash, _MessageEq>
+      sending_queue_;
+  // Keep track of the versions of components in this node.


I'm still a bit confused: is this about the latest version the local node has received or sent?

fishbone · 2022-03-28T22:37:07Z

Could you update the comment then? The comment is saying handling updates send from this node.

Here this node means the node associated with the connection.

fishbone · 2022-03-28T22:43:36Z

@jjyao I think you mistake the connection and the local node here. Connection is about the node connected to the local node. All node_id in the connection is about the node connected to the local node.

We went through the code in the morning. Do you mind revisiting your comment and trying to see which one still makes sense?

src/ray/common/ray_syncer/ray_syncer-inl.h

+
+  RaySyncer &instance_;
+  instrumented_io_context &io_context_;
+  std::string node_id_;


src/ray/common/ray_syncer/ray_syncer-inl.h

+
+  absl::flat_hash_set<std::shared_ptr<const RaySyncMessage>, _MessageHash, _MessageEq>
+      sending_queue_;
+  // Keep track of the versions of components in this node.


jjyao · 2022-03-29T03:39:28Z

src/ray/common/ray_syncer/ray_syncer.cc

+    if (node_versions[message.component_id()] < message.version()) {
+      node_versions[message.component_id()] = message.version();
+    }
+    instance_.BroadcastMessage(std::make_shared<RaySyncMessage>(std::move(message)));


Do we need to broadcast in the else case? Doesn't that mean we received an old message that should be discarded?

I agree with you.

src/ray/common/ray_syncer/ray_syncer.h

fishbone · 2022-03-29T05:25:21Z

Thanks, @jjyao for the comments. I'll update the PR.

jjyao

LGTM

jjyao · 2022-03-29T04:42:48Z

src/ray/common/ray_syncer/ray_syncer.cc

+            if (!message) {
+              continue;
+            }
+            if (upward_only_[message->component_id()] && !is_upward_conn) {


What's the example of upward_only component?

I put it in the Register comments.

jjyao · 2022-03-29T04:45:58Z

src/ray/common/ray_syncer/ray_syncer.cc

+}
+
+bool NodeState::ConsumeMessage(std::shared_ptr<const RaySyncMessage> message) {
+  auto &current = cluster_view_[message->node_id()][message->component_id()];


Do we need to check whether message->node_id() == node_id and skip if so?

I do it on purpose. 1) if there is a new connection, sending the snapshot will be easy to do. 2) receiver might want to handle the local updates and this gives a way for the local reporter communicates to the local receiver.

src/ray/common/ray_syncer/ray_syncer.cc

src/ray/common/ray_syncer/ray_syncer.h

scv119 · 2022-03-29T06:35:31Z

src/ray/common/ray_syncer/ray_syncer-inl.h

+      cluster_view_;
+};
+
+class NodeSyncConnection {


think again about this NodeSyncConnection. If we are using it to simulating streaming, should it have following APIs?

class NodeSyncConnection { // Receives stream of responses from remote node. using ResponseObserver = std::function<void(Response)>; public: NodeSyncConnection(ResponseObserver response_observer); // Sends requests to remote node. bool SendRequest(Request request); };

@scv119 even with streaming, we still need Server and Client concept here. The one who initializes the work is client.

Here the difference is: 1) initialization; 2) how to send; 3) how to receive. All these are communication layer things.

Put them into NodeSyncConnection gives me the feeling that it's not a good abstraction.

Now it's like:

NodeSyncConnection -> application layer filtering

Server/ClientSyncConneciton-> Handling how to do communication.

hmm could you hide them under the NodeSyncConnection interface/abstract class?

class NodeSyncConnection { // Receives stream of responses from remote node. using ResponseObserver = std::function<void(Response)>; public: NodeSyncConnection(ResponseObserver response_observer); // Sends requests to remote node. bool SendRequest(Request request); }; class ServerSyncConnection : public NodeSyncConnection { public: ServerSyncConnection(ResponseObserver response_observer); bool SendRequest(Request request) override; }; class ClientSyncConnection : public NodeSyncConnection { public: ClientSyncConnection(ResponseObserver response_observer); bool SendRequest(Request request) override; };

I think it's not changing anything.

PushToSendingQueue is SendRequest

ReceiveUpdates is ResponseObserver

Btw, the only different part is about communication, what to do when we receive the updates is the same.

And we still end up having two subclasses of NodeSyncConnection.

Once, thing I think we probably can do is DoSend is a callback. But is this better than the current implementation?

I just updated it. I think it looks better than before.

fishbone · 2022-03-29T20:15:47Z

@scv119 @rkooo567 could you give another review for this PR?

fishbone · 2022-03-30T03:36:58Z

Discussed with @scv119 offline and I'll move ray syncer out of NodeSyncConnection.

fishbone · 2022-03-30T06:51:44Z

Mac is hanging forever ... and the failure is not related.

fishbone · 2022-03-30T06:53:58Z

Thanks @scv119 @jjyao @rkooo567 who help reviewed this PR. I feel along with the discussion and iterating, the PR has been better than the first one. Really appreciate your time!

fishbone force-pushed the refactor-syncer branch from 8c274e7 to 0fe7d0f Compare March 17, 2022 07:28

initial version

91667ff

fishbone force-pushed the refactor-syncer branch from 0fe7d0f to 91667ff Compare March 17, 2022 07:30

add more comments

0b3265c

fishbone changed the title ~~Refactor syncer~~ [wip][scheduling][5] Refactor resource syncer. Mar 17, 2022

fishbone marked this pull request as ready for review March 17, 2022 07:42

add proto comment

f47727e

scv119 reviewed Mar 17, 2022

View reviewed changes

src/ray/protobuf/ray_syncer.proto Outdated Show resolved Hide resolved

scv119 reviewed Mar 17, 2022

View reviewed changes

src/ray/protobuf/ray_syncer.proto Outdated Show resolved Hide resolved

scv119 reviewed Mar 17, 2022

View reviewed changes

src/ray/protobuf/ray_syncer.proto Outdated Show resolved Hide resolved

scv119 reviewed Mar 17, 2022

View reviewed changes

src/ray/protobuf/ray_syncer.proto Outdated Show resolved Hide resolved

scv119 reviewed Mar 17, 2022

View reviewed changes

src/ray/common/ray_syncer.h Outdated Show resolved Hide resolved

scv119 reviewed Mar 17, 2022

View reviewed changes

src/ray/common/ray_syncer.h Outdated Show resolved Hide resolved

scv119 reviewed Mar 17, 2022

View reviewed changes

src/ray/common/ray_syncer.h Outdated Show resolved Hide resolved

fishbone and others added 8 commits March 17, 2022 17:56

update

f386753

update

9a5ffbc

update

e870bd6

Merge branch 'refactor-syncer' of github.com:iycheng/ray into refacto…

6d6989f

…r-syncer

add test_syncer_service

21dd1e4

format

46ca5aa

make example compilable

700aead

format

b7111f0

fishbone commented Mar 17, 2022

View reviewed changes

src/ray/common/ray_syncer.h Outdated Show resolved Hide resolved

fishbone commented Mar 17, 2022

View reviewed changes

fishbone added 2 commits March 18, 2022 08:50

update

5d17d83

format

b4aa776

scv119 reviewed Mar 18, 2022

View reviewed changes

fix comment

958e71a

fishbone added 3 commits March 25, 2022 07:05

try to fix tsan

da5b74f

Merge remote-tracking branch 'upstream/master' into refactor-syncer

b3cb08d

another try to fix tsan

c71d847

jjyao reviewed Mar 25, 2022

View reviewed changes

fishbone added 3 commits March 28, 2022 00:52

try to fix flaky

0c20810

format

cf8e28f

a try to fix

7a879b5

fishbone force-pushed the refactor-syncer branch from 331085a to 7a879b5 Compare March 28, 2022 08:56

fishbone added 3 commits March 28, 2022 18:45

fix

ef6c28b

fix

85fc7fb

Merge remote-tracking branch 'upstream/master' into refactor-syncer

6b880ea

fix some

b654412

jjyao reviewed Mar 29, 2022

View reviewed changes

fishbone added 2 commits March 29, 2022 06:19

fix one bug!

498c335

format

4a6cb59

scv119 reviewed Mar 29, 2022

View reviewed changes

fix comments

e3aa592

jjyao approved these changes Mar 29, 2022

View reviewed changes

scv119 approved these changes Mar 29, 2022

View reviewed changes

fix comment

16b5eab

fishbone merged commit 781c46a into ray-project:master Mar 30, 2022

fishbone mentioned this pull request Apr 26, 2022

[scheduler][6] Integrate ray with syncer. #23660

Merged

6 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[scheduling][5] Refactor resource syncer. #23270

[scheduling][5] Refactor resource syncer. #23270

fishbone commented Mar 17, 2022 •

edited

Loading

fishbone commented Mar 17, 2022

fishbone left a comment

fishbone commented Mar 18, 2022 •

edited

Loading

scv119 Mar 18, 2022 •

edited

Loading

fishbone Mar 18, 2022

jjyao left a comment

jjyao Mar 25, 2022

fishbone Mar 28, 2022

This comment was marked as resolved.

jjyao Mar 25, 2022

jjyao Mar 25, 2022

fishbone commented Mar 28, 2022

fishbone commented Mar 28, 2022 •

edited

Loading

This comment was marked as resolved.

This comment was marked as resolved.

jjyao Mar 29, 2022

fishbone Mar 29, 2022

fishbone commented Mar 29, 2022

jjyao left a comment

jjyao Mar 29, 2022

fishbone Mar 29, 2022

jjyao Mar 29, 2022

fishbone Mar 29, 2022

scv119 Mar 29, 2022

fishbone Mar 29, 2022

scv119 Mar 29, 2022 •

edited

Loading

fishbone Mar 29, 2022

fishbone Mar 30, 2022

fishbone commented Mar 29, 2022

fishbone commented Mar 30, 2022

fishbone commented Mar 30, 2022

fishbone commented Mar 30, 2022

		/// It'll send nothing unless there is a request from the remote node
		/// for the sending request.

[scheduling][5] Refactor resource syncer. #23270

[scheduling][5] Refactor resource syncer. #23270

Conversation

fishbone commented Mar 17, 2022 • edited Loading

Why are these changes needed?

Related issue number

Checks

fishbone commented Mar 17, 2022

fishbone left a comment

Choose a reason for hiding this comment

fishbone commented Mar 18, 2022 • edited Loading

scv119 Mar 18, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jjyao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

This comment was marked as resolved.

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fishbone commented Mar 28, 2022

fishbone commented Mar 28, 2022 • edited Loading

This comment was marked as resolved.

This comment was marked as resolved.

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fishbone commented Mar 29, 2022

jjyao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scv119 Mar 29, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fishbone commented Mar 29, 2022

fishbone commented Mar 30, 2022

fishbone commented Mar 30, 2022

fishbone commented Mar 30, 2022

fishbone commented Mar 17, 2022 •

edited

Loading

fishbone commented Mar 18, 2022 •

edited

Loading

scv119 Mar 18, 2022 •

edited

Loading

fishbone commented Mar 28, 2022 •

edited

Loading

scv119 Mar 29, 2022 •

edited

Loading