feat(meta): inject and collect barrier in bidi stream #14887

wenym1 · 2024-01-31T03:02:25Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

In this PR, we replace the inject_barrier, collect_barrier and force_stop_actors rpc with a bidi-stream called StreamingControlStream.

When a new stream between meta and CN is established, the CN will be reset, which is similar to the original force_stop_actors. After the stream is established, meta node will inject barrier via a InjectBarrierRequest request in the stream, and when the barrier is collected on CN, CN will response with a BarrierCompleteResponse. InjectBarrierResponse and BarrierCompleteRequest is no longer needed and is removed in this PR.

message StreamingControlStreamRequest {
  message InitRequest {
    uint64 prev_epoch = 2;
  }

  oneof request {
    InitRequest init = 1;
    InjectBarrierRequest inject_barrier = 2;
  }
}

message StreamingControlStreamResponse {
  message InitResponse {}

  oneof response {
    InitResponse init = 1;
    BarrierCompleteResponse complete_barrier = 2;
  }
}

rpc StreamingControlStream(stream StreamingControlStreamRequest) returns (stream StreamingControlStreamResponse);

Any CN failure, such as failure to collect a barrier or CN crashes, will be reflected to meta node as an error on the stream. On error, the meta node will trigger a recovery, which will reestablish the stream between meta and each CN, and each CN will be reset.

On meta side, in CheckpointControl previously we have an enum BarrierEpochState to represent the state of barrier, which can be either Inflight, which means the barrier is injected but not yet completed, or Completed, which means the barrier has been collected from all CNs. In this PR, the barrier may have been collected from a subset of CNs, and there are some pending CNs to be collect, and therefore we change it to the following struct to store the node to be collected and the collected response. When node_to_collect gets empty, it means the barrier has been collected from all CNs, and we can continue complete the barrier.

struct BarrierEpochState {
    node_to_collect: HashSet<WorkerId>,
    resps: Vec<BarrierCompleteResponse>,
}

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

gitguardian · 2024-02-19T07:41:54Z

️✅ There are no secrets present in this pull request anymore.

If these secrets were true positive and are still valid, we highly recommend you to revoke them.
Once a secret has been leaked into a git repository, you should consider it compromised, even if it was deleted immediately.
Find here more information about risks.

^{_{🦉 GitGuardian detects secrets in your source code to help developers and security teams secure the modern development process. You are seeing this because you or someone else with access to this repository has authorized GitGuardian to scan your pull request.

Our GitHub checks need improvements? Share your feedbacks!}}

kwannoel · 2024-02-27T05:16:06Z

Can you add some comments on the motivation behind this PR?

wenym1 · 2024-03-05T08:50:18Z

Can you add some comments on the motivation behind this PR?

The main motivation behind this PR is for support of partial checkpoint. In the design of partial checkpoint, barrier collection will be reported to meta in the granularity of per MV or per actor, and there will be more control messages sent between meta and CN. Therefore it's not suitable to use a single rpc to collect barrier from all actors any more.

Apart from support for partial checkpoint, there is some general benefits replace oneshot rpc calls with a single bidi stream.

First, we can avoid out of order messages. For example, currently when injecting barrier, we need to wait for the inject response before injecting the next barrier because inflight messages of multiple concurrents rpc can be out of order. With bidi stream, the messages must be in the order of being sent.

Second, we can avoid stale requests. In current code, some stale rpc might be still in flight when recovery happens, and may be handled after recovery, and we will need extra effort to handle these stale requests. After this PR, a new control stream will be established in recovery and all previously in flight request in the stream will be discarded, and we can get rid of stale requests.

wenym1 · 2024-03-05T09:12:10Z

@hzxa21 @BugenZhao @kwannoel @yezizp2012 @shanicky The PR is ready for review now. PTAL~

MrCroxx

Could you please explain why introducing a bidi-stream to solve the problem? What's the limitation of the original design? Thanks.

hzxa21

LGTM!

hzxa21 · 2024-03-07T08:18:34Z

src/meta/src/barrier/recovery.rs

+                    self.context.update_actors(&info).await.inspect_err(|err| {
                        warn!(error = %err.as_report(), "update actors failed");
                    })?;
-                    self.build_actors(&info).await.inspect_err(|err| {
+                    self.context.build_actors(&info).await.inspect_err(|err| {


Just curious: will it be beneficial if we move update_actors and build_actors into the bidi-streaming rpc?

src/compute/src/rpc/service/stream_service.rs

kwannoel · 2024-03-07T13:39:05Z

src/meta/src/barrier/command.rs

@@ -175,6 +175,8 @@ pub enum Command {
    RescheduleFragment {
        reschedules: HashMap<FragmentId, Reschedule>,
        table_parallelism: HashMap<TableId, TableParallelism>,
+        // should contain the actor ids in upstream and downstream fragment of `reschedules`
+        fragment_actors: HashMap<FragmentId, HashSet<ActorId>>,


Minor comment, I suppose RescheduleFragment is actually RescheduleFragments? Hence we have a hashmap here, rather than just the actor ids in upstream and downstream fragment of reschedules.

Also why we refactor and add the field fragment_actors here?

I guess it is to make to_mutation non-async

Yes. It's for making it non-async.

BugenZhao · 2024-03-08T04:20:45Z

src/meta/src/barrier/mod.rs

@@ -489,7 +496,7 @@ impl GlobalBarrierManager {
    }

    /// Check whether we should pause on bootstrap from the system parameter and reset it.
-    async fn take_pause_on_bootstrap(&self) -> MetaResult<bool> {
+    async fn take_pause_on_bootstrap(&mut self) -> MetaResult<bool> {


What's this change for?

If we take &self, the return future will take the immutable reference to self. The future is Send only when &self is Send, which then requires self to be Sync. However, after this PR, self holds the bidi stream returned from rpc client, which is only Send but not Sync. If we don't change it, we will fail to compile it.

Interesting. This sounds like the problem that Exclusive is designed for.

src/meta/src/barrier/rpc.rs

BugenZhao · 2024-03-08T04:31:36Z

src/meta/src/barrier/rpc.rs

+type ResponseStreamFuture = impl Future<
+        Output = (
+            WorkerId,
+            BoxStream<
+                'static,
+                risingwave_rpc_client::error::Result<StreamingControlStreamResponse>,
+            >,
+            MetaResult<StreamingControlStreamResponse>,
+        ),
+    > + 'static;


Wow, so complicated. 🫨

src/rpc_client/src/lib.rs

src/stream/src/task/stream_manager.rs

BugenZhao · 2024-03-08T04:39:39Z

InjectBarrierResponse and BarrierCompleteRequest is no longer needed and is removed in this PR.

Prior to this PR, we have await-tree instrumentation for all RPC calls. The tree for barrier_complete could be useful for developers to investigate the barriers piling up on a specific compute node. Is this still available now?

wenym1 · 2024-03-08T12:49:16Z

InjectBarrierResponse and BarrierCompleteRequest is no longer needed and is removed in this PR.

Prior to this PR, we have await-tree instrumentation for all RPC calls. The tree for barrier_complete could be useful for developers to investigate the barriers piling up on a specific compute node. Is this still available now?

It's not supported in the PR currently. Do you have any recommendation on how to support it? Is it doable to store a await tree tracing span in the inflight barrier queue?

BugenZhao · 2024-03-11T03:58:34Z

InjectBarrierResponse and BarrierCompleteRequest is no longer needed and is removed in this PR.

Prior to this PR, we have await-tree instrumentation for all RPC calls. The tree for barrier_complete could be useful for developers to investigate the barriers piling up on a specific compute node. Is this still available now?

It's not supported in the PR currently. Do you have any recommendation on how to support it? Is it doable to store a await tree tracing span in the inflight barrier queue?

I don't have an idea. #4581 introduces a middleware for non-streaming gRPC calls so that they can be instrumented transparently. Now it seems that we have to find a way to do that manually.

wenym1 · 2024-03-11T08:50:35Z

InjectBarrierResponse and BarrierCompleteRequest is no longer needed and is removed in this PR.

Prior to this PR, we have await-tree instrumentation for all RPC calls. The tree for barrier_complete could be useful for developers to investigate the barriers piling up on a specific compute node. Is this still available now?

It's not supported in the PR currently. Do you have any recommendation on how to support it? Is it doable to store a await tree tracing span in the inflight barrier queue?

I don't have an idea. #4581 introduces a middleware for non-streaming gRPC calls so that they can be instrumented transparently. Now it seems that we have to find a way to do that manually.

Split to Implement in #15608, given that the current PR is quite large.

This reverts commit 72d37ef.

…" (#15652)

…5653)

…" (#15652)

…5653)

github-actions bot added the type/feature label Jan 31, 2024

wenym1 marked this pull request as ready for review February 21, 2024 03:18

wenym1 requested review from shanicky, hzxa21, yezizp2012, BugenZhao and kwannoel February 26, 2024 12:22

feat(meta): inject and collect barrier in bidi stream

9bf3366

wenym1 force-pushed the yiming/barrier-bidi-stream branch from e2e5807 to 9bf3366 Compare March 5, 2024 08:34

fix test

39d763f

wenym1 requested a review from MrCroxx March 6, 2024 05:35

Merge branch 'main' into yiming/barrier-bidi-stream

f554c62

MrCroxx reviewed Mar 7, 2024

View reviewed changes

hzxa21 approved these changes Mar 7, 2024

View reviewed changes

kwannoel reviewed Mar 7, 2024

View reviewed changes

src/compute/src/rpc/service/stream_service.rs Outdated Show resolved Hide resolved

kwannoel reviewed Mar 7, 2024

View reviewed changes

BugenZhao reviewed Mar 8, 2024

View reviewed changes

fix comment

e0f5aa0

Merge branch 'main' into yiming/barrier-bidi-stream

2f854ff

wenym1 added this pull request to the merge queue Mar 12, 2024

Merged via the queue into main with commit 72d37ef Mar 12, 2024
28 of 29 checks passed

wenym1 deleted the yiming/barrier-bidi-stream branch March 12, 2024 07:19

lmatz mentioned this pull request Mar 13, 2024

nightly-20240312 pipeline test failed #15650

Closed

TennyZhuang added a commit that referenced this pull request Mar 13, 2024

Revert "feat(meta): inject and collect barrier in bidi stream (#14887)"

a82597d

This reverts commit 72d37ef.

TennyZhuang mentioned this pull request Mar 13, 2024

revert: "feat(meta): inject and collect barrier in bidi stream (#14887)" #15652

Merged

9 tasks

TennyZhuang added a commit that referenced this pull request Mar 13, 2024

revert: "feat(meta): inject and collect barrier in bidi stream (#14887)…

59ae1cf

…" (#15652)

wenym1 mentioned this pull request Mar 13, 2024

feat: re-merge #14887 (inject and collect barrier in bidi stream) #15653

Merged

github-merge-queue bot pushed a commit that referenced this pull request Mar 13, 2024

feat: re-merge #14887 (inject and collect barrier in bidi stream) (#1…

e46a307

…5653)

jetjinser pushed a commit that referenced this pull request Mar 14, 2024

feat(meta): inject and collect barrier in bidi stream (#14887)

171a37e

jetjinser pushed a commit that referenced this pull request Mar 14, 2024

revert: "feat(meta): inject and collect barrier in bidi stream (#14887)…

14fa428

…" (#15652)

jetjinser pushed a commit that referenced this pull request Mar 14, 2024

feat: re-merge #14887 (inject and collect barrier in bidi stream) (#1…

5b159f0

…5653)

BugenZhao mentioned this pull request Mar 15, 2024

e2e parallel tests failed: Failed to send barrier with epoch 6107819919474688 to actor 9859: sender not found #15703

Closed

wenym1 mentioned this pull request Mar 19, 2024

Tracking: support partial checkpoint #14041

Open

25 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(meta): inject and collect barrier in bidi stream #14887

feat(meta): inject and collect barrier in bidi stream #14887

wenym1 commented Jan 31, 2024 •

edited

Loading

gitguardian bot commented Feb 19, 2024 •

edited

Loading

kwannoel commented Feb 27, 2024

wenym1 commented Mar 5, 2024

wenym1 commented Mar 5, 2024

MrCroxx left a comment

hzxa21 left a comment

hzxa21 Mar 7, 2024

kwannoel Mar 7, 2024

kwannoel Mar 7, 2024

hzxa21 Mar 8, 2024

wenym1 Mar 8, 2024

BugenZhao Mar 8, 2024

wenym1 Mar 8, 2024

BugenZhao Mar 11, 2024

BugenZhao Mar 8, 2024

BugenZhao commented Mar 8, 2024

wenym1 commented Mar 8, 2024

BugenZhao commented Mar 11, 2024

wenym1 commented Mar 11, 2024

feat(meta): inject and collect barrier in bidi stream #14887

feat(meta): inject and collect barrier in bidi stream #14887

Conversation

wenym1 commented Jan 31, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

gitguardian bot commented Feb 19, 2024 • edited Loading

️✅ There are no secrets present in this pull request anymore.

kwannoel commented Feb 27, 2024

wenym1 commented Mar 5, 2024

wenym1 commented Mar 5, 2024

MrCroxx left a comment

Choose a reason for hiding this comment

hzxa21 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BugenZhao commented Mar 8, 2024

wenym1 commented Mar 8, 2024

BugenZhao commented Mar 11, 2024

wenym1 commented Mar 11, 2024

wenym1 commented Jan 31, 2024 •

edited

Loading

gitguardian bot commented Feb 19, 2024 •

edited

Loading