feat(cdc-backfill): support pause and resume #14590
Conversation
Signed-off-by: TennyZhuang <[email protected]>
TennyZhuang
force-pushed
the
cdc-backfill-pause
branch
from
2a2bf40 to
0000002
Compare
| #[auto_enum(futures03::Stream)] | ||
| let right_snapshot = if !paused { | ||
| upstream_table_reader.snapshot_read(args).map(Either::Right) | ||
| } else { | ||
| futures::stream::pending() | ||
| }; |
There was a problem hiding this comment.
What about
risingwave/src/stream/src/executor/stream_reader.rs
Lines 33 to 48 in a5e510c
IIRC, pause_on_startup flag will be reset to false after meta restart. So in the second time we restart meta, it will be resumed. Anything I missed? @BugenZhao |
It can also be resumed by issuing a |
Yep, this pr doesn't support this resume path. |
|
Reimplemented and supported resume, please review again. |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
license-eye has totally checked 4727 files.
| Valid | Invalid | Ignored | Fixed |
|---|---|---|---|
| 2065 | 1 | 2661 | 0 |
Click to see the invalid file list
- src/utils/futures_util/src/pausable.rs
Signed-off-by: TennyZhuang <[email protected]>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
| fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { | ||
| let this = self.project(); | ||
| if this.paused.load(Ordering::Relaxed) { | ||
| Poll::Pending |
There was a problem hiding this comment.
Returning Pending without doing anything with the waker could potentially lead to deadlock.
There was a problem hiding this comment.
Can you give me more explanation? I can only find https://docs.rs/futures-util/0.3.30/src/futures_util/stream/pending.rs.html#32
There was a problem hiding this comment.
Found the article, but I can't find the solution.
There was a problem hiding this comment.
Oh, should I register the paused as a signal?
There was a problem hiding this comment.
Here's a simple reproduction. 🥹
There was a problem hiding this comment.
Oh, should I register the paused as a signal?
Yeah. I guess using anything else with a wait-queue mechanism should work.
Although I guess the deadlock won't occur with the usage of this PR, but since it's made a general utility, we should be more careful about it.
Signed-off-by: TennyZhuang <[email protected]>
Signed-off-by: TennyZhuang <[email protected]>
Signed-off-by: TennyZhuang <[email protected]>
Signed-off-by: TennyZhuang <[email protected]>
TennyZhuang
changed the title
| let this = self.project(); | ||
| if this.paused.load(Ordering::Relaxed) { | ||
| let mut waker = this.waker.lock().unwrap(); | ||
| waker.get_or_insert(cx.waker().clone()); |
There was a problem hiding this comment.
Should we always insert the new waker? Is it possible that the stream is moved to another task and the previous waker does not work anymore?
There was a problem hiding this comment.
I'm unsure, but insert seems not bad here. Updated
Signed-off-by: TennyZhuang <[email protected]>
Signed-off-by: TennyZhuang <[email protected]>
Signed-off-by: TennyZhuang <[email protected]> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
#14555
Checklist
./risedev check(or alias,./risedev c)Documentation
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.