[fix #241] fix flow control #242

zeminzhou · 2022-09-28T07:19:25Z

Signed-off-by: zeminzhou [email protected]

What problem does this PR solve?

Fix flow control

Issue Number: close #241

Problem Description: TBD

What is changed and how does it work?

Fix flow control

Code changes

Has exported function/method change

Check List for Tests

This PR has been tested by at least one of the following methods:

Unit test
Integration test
Manual test (add detailed scripts or steps below)

Signed-off-by: zeminzhou <[email protected]>

zeminzhou · 2022-09-28T07:21:59Z

/cc @pingyu @haojinming PTAL~

codecov · 2022-09-28T07:34:28Z

Codecov Report

Merging #242 (9f4a2bf) into main (bf349c5) will decrease coverage by 0.3052%.
The diff coverage is 6.8181%.

Additional details and impacted files

@@               Coverage Diff                @@
##               main       #242        +/-   ##
================================================
- Coverage   61.3330%   61.0277%   -0.3053%     
================================================
  Files           239        239                
  Lines         20180      20199        +19     
================================================
- Hits          12377      12327        -50     
- Misses         6666       6755        +89     
+ Partials       1137       1117        -20

Flag	Coverage Δ		*Carryforward flag
br	`60.2203% <ø> (ø)`		Carriedforward from 8d809a8
cdc	`61.3984% <6.8181%> (-0.4461%)`	⬇️

*This pull request uses carry forward flags. Click here to find out more.

Impacted Files	Coverage Δ
cdc/cdc/processor/pipeline/keyspan.go	`0.0000% <0.0000%> (ø)`
cdc/cdc/processor/pipeline/metrics.go	`0.0000% <0.0000%> (ø)`
cdc/cdc/processor/pipeline/sorter.go	`30.8333% <0.0000%> (-5.4412%)`	⬇️
cdc/cdc/sink/common/flow_control.go	`0.0000% <0.0000%> (ø)`
cdc/cdc/processor/pipeline/sink.go	`63.0434% <100.0000%> (ø)`
cdc/pkg/config/server_config.go	`55.6818% <100.0000%> (ø)`
cdc/cdc/sorter/unified/file_backend.go	`33.7719% <0.0000%> (-20.1755%)`	⬇️
cdc/pkg/pdtime/acquirer.go	`58.1395% <0.0000%> (-6.9768%)`	⬇️
cdc/pkg/orchestrator/etcd_worker.go	`78.5714% <0.0000%> (-0.8404%)`	⬇️
cdc/cdc/sorter/unified/unified_sorter.go	`91.6666% <0.0000%> (-0.6945%)`	⬇️
... and 1 more

haojinming · 2022-09-28T07:39:41Z

cdc/cdc/processor/pipeline/sorter.go

+								zap.Uint64("keyspanID", n.keyspanID),
+								zap.String("keyspanName", n.keyspanName))
+						} else {
+							ctx.Throw(err)


Does this cause panic?

It does not cause panic, just throws a err to parents ndoes.

pingyu

Suggest to add some test case to verify the consuming & releasing (including integrated & e2e tests). If we miss some release, the flowController.Consume will block the data flow.

pingyu · 2022-10-08T13:31:09Z

cdc/cdc/processor/pipeline/sorter.go

@@ -188,6 +212,7 @@ func (n *sorterNode) TryHandleDataMessage(ctx context.Context, msg pipeline.Mess
 					zap.Uint64("resolvedTs", resolvedTs),
 					zap.Uint64("oldResolvedTs", oldResolvedTs))
 			}
+			atomic.StoreUint64(&n.resolvedTs, rawKV.CRTs)


Why ? The previous oldResolvedTs := atomic.SwapUint64(&n.resolvedTs, resolvedTs) should have done the assignment.

Removed, thanks! PTAL~

pingyu · 2022-10-08T13:38:02Z

cdc/cdc/processor/pipeline/sorter.go

@@ -151,6 +151,30 @@ func (n *sorterNode) StartActorNode(ctx pipeline.NodeContext, eg *errgroup.Group
 						}
 					}

+					// We calculate memory consumption by PolymorphicEvent size.


About PerKeySpanMemoryQuota:

The default value 10MB may be too small. 1GB may be fine.

I think we can change the config item name per-keyspan-memory-quota to per-changefeed-memory-quota, as users have no idea that what keyspan is.

It would better to default to a proportion of total system memory, for ease of use. Refer to MaxMemoryPresure. (Not a must in this PR).

Thanks for your suggestion, I do the following:

changed 10MB to 1GB;

changed per-keyspan-memory-quota to per-changefeed-memory-quota;

Signed-off-by: zeminzhou <[email protected]>

haojinming · 2022-10-10T02:58:38Z

cdc/cdc/processor/pipeline/sorter.go

+					size := uint64(msg.RawKV.ApproximateDataSize())
+					// NOTE we allow the quota to be exceeded if blocking means interrupting a transaction.
+					// Otherwise the pipeline would deadlock.
+					err := n.flowController.Consume(commitTs, size, func() error {


Suggest add a metric to observe the blocking duration and Consume failure count. It seems it should not fail unless flowcontrol is aborted.

added, thanks! PTAL~

Signed-off-by: zeminzhou <[email protected]>

zeminzhou · 2022-10-10T07:23:38Z

Signed-off-by: zeminzhou <[email protected]>

haojinming

LGTM

pingyu

Rest LGTM.

pingyu · 2022-10-10T14:58:30Z

cdc/cdc/processor/pipeline/metrics.go

 			Help:      "estimated memory consumption for a keyspan after the sorter",
 			Buckets:   prometheus.ExponentialBuckets(1*1024*1024 /* mb */, 2, 10),
 		}, []string{"changefeed", "capture"})
+	flowControllerConsumeHistogram = prometheus.NewHistogramVec(


Suggest to name as flowControllerDurationHistogram and add a tag type = consume for Consume interface. Then we can add more tags for other interfaces if necessary later.

added, thanks! PTAL~

pingyu · 2022-10-10T15:09:17Z

cdc/metrics/grafana/tikv-cdc.json

@@ -2965,7 +2965,7 @@
          "steppedLine": false,
          "targets": [
            {
-              "expr": "sum(rate(tikv_cdc_processor_table_memory_consumption_sum{tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"}[30s]) / rate(tikv_cdc_processor_table_memory_consumption_count{tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"}[30s])) by (capture)",
+              "expr": "sum(rate(tikv_cdc_processor_changefeed_memory_consumption_sum{tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"}[30s]) / rate(tikv_cdc_processor_changefeed_memory_consumption_count{tidb_cluster=\"$tidb_cluster\", capture=~\"$capture\"}[30s])) by (capture)",


Please also paste the screenshot of these two metrics.

pasted, PTAL -> #242 (comment)

pingyu · 2022-10-10T15:12:12Z

/verify

Signed-off-by: zeminzhou <[email protected]>

pingyu

LGTM~

zeminzhou added 2 commits September 27, 2022 14:14

fix flow control

f2ac417

Signed-off-by: zeminzhou <[email protected]>

remove barrier

8b5c26f

Signed-off-by: zeminzhou <[email protected]>

zeminzhou requested review from pingyu and haojinming September 28, 2022 07:22

haojinming reviewed Sep 28, 2022

View reviewed changes

pingyu reviewed Oct 8, 2022

View reviewed changes

zeminzhou added 5 commits October 9, 2022 14:51

Merge branch 'main' into fix-flow-control

bf2f05d

KeySpanFlowController -> ChangefeedFlowController

8a447b7

Signed-off-by: zeminzhou <[email protected]>

fix ut

f6bc4b5

Signed-off-by: zeminzhou <[email protected]>

Merge branch 'main' into fix-flow-control

94206fe

fix ut

55e85cf

Signed-off-by: zeminzhou <[email protected]>

haojinming reviewed Oct 10, 2022

View reviewed changes

zeminzhou added 5 commits October 10, 2022 11:03

rename NewKeySpanMemoryQuota -> NewChangefeedMemoryQuota

acaf9b1

Signed-off-by: zeminzhou <[email protected]>

fix ut

5c4b4f2

Signed-off-by: zeminzhou <[email protected]>

fix ut

fa25017

Signed-off-by: zeminzhou <[email protected]>

fix metric for flow-control

e1ab901

Signed-off-by: zeminzhou <[email protected]>

fix metric for flow-control

f7e830a

Signed-off-by: zeminzhou <[email protected]>

add metric for flowController consume

40c605a

Signed-off-by: zeminzhou <[email protected]>

haojinming approved these changes Oct 10, 2022

View reviewed changes

pingyu reviewed Oct 10, 2022

View reviewed changes

zeminzhou added 2 commits October 11, 2022 11:08

fix comment

8d809a8

Signed-off-by: zeminzhou <[email protected]>

fix comment

9f4a2bf

Signed-off-by: zeminzhou <[email protected]>

pingyu approved these changes Oct 11, 2022

View reviewed changes

pingyu merged commit 21754a5 into tikv:main Oct 11, 2022

zeminzhou deleted the fix-flow-control branch October 17, 2022 13:20

haojinming mentioned this pull request Feb 9, 2023

CDC does not sink data to destination tikv cluster #161

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix #241] fix flow control #242

[fix #241] fix flow control #242

zeminzhou commented Sep 28, 2022

zeminzhou commented Sep 28, 2022 •

edited

Loading

codecov bot commented Sep 28, 2022 •

edited

Loading

haojinming Sep 28, 2022

zeminzhou Oct 10, 2022

pingyu left a comment

pingyu Oct 8, 2022

zeminzhou Oct 10, 2022

pingyu Oct 8, 2022

zeminzhou Oct 10, 2022

haojinming Oct 10, 2022 •

edited

Loading

zeminzhou Oct 10, 2022

zeminzhou commented Oct 10, 2022

haojinming left a comment

pingyu left a comment

pingyu Oct 10, 2022

zeminzhou Oct 11, 2022

pingyu Oct 10, 2022

zeminzhou Oct 11, 2022

pingyu commented Oct 10, 2022

pingyu left a comment

[fix #241] fix flow control #242

[fix #241] fix flow control #242

Conversation

zeminzhou commented Sep 28, 2022

What problem does this PR solve?

What is changed and how does it work?

Code changes

Check List for Tests

zeminzhou commented Sep 28, 2022 • edited Loading

codecov bot commented Sep 28, 2022 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pingyu 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

haojinming Oct 10, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zeminzhou commented Oct 10, 2022

haojinming left a comment

Choose a reason for hiding this comment

pingyu 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

pingyu commented Oct 10, 2022

pingyu left a comment

Choose a reason for hiding this comment

zeminzhou commented Sep 28, 2022 •

edited

Loading

codecov bot commented Sep 28, 2022 •

edited

Loading

haojinming Oct 10, 2022 •

edited

Loading