relay/syncer: relay notifies syncer of new write to reduce sync latency #2225

D3Hunter · 2021-10-15T04:11:18Z

What problem does this PR solve?

subtask 2 and 3 of relay: refactor dm-worker relay logic #2214
fix unstable integration test case: ha_cases3(see comments below)

What is changed and how it works?

add Listener function to relay, it notifies after writing events successfully
add EventNotifier to local binlog reader(used by syncer), and change polling file change to waiting on write events to reduce sync latency

Check List

Tests

Unit test

ti-chi-bot · 2021-10-15T04:11:19Z

[REVIEW NOTIFICATION]

This pull request has been approved by:

Ehco1996
lichunzhu

To complete the pull request process, please ask the reviewers in the list to review by filling /cc @reviewer in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the list by filling /assign @committer in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found here.

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.

D3Hunter · 2021-10-15T04:23:52Z

/cc @lichunzhu

dm/worker/relay.go

dm/worker/subtask.go

dm/worker/subtask_holder.go

relay/relay.go

syncer/syncer.go

relay/relay.go

loader/lightning.go

pkg/streamer/file.go

relay/relay.go

pkg/streamer/reader.go

syncer/syncer.go

Ehco1996 · 2021-10-19T00:48:41Z

dm/worker/subtask_holder.go

+// OnEvent implements relay.Listener
+// only syncer unit of subtask need to be notified, but it's much simpler and less error-prone to manage it here
+// as relay event need to broadcast to every syncer(most subtask have a syncer).
+func (h *subTaskHolder) OnEvent(e *replication.BinlogEvent) {


seems e *replication.BinlogEvent is not used here, why we need this param ?

it's a event listener
the parameter is not used now, but it's added for completeness(observer pattern), may used later.

Suggested change

func (h *subTaskHolder) OnEvent(e *replication.BinlogEvent) {

func (h *subTaskHolder) OnEvent(_ *replication.BinlogEvent) {

pkg/streamer/file_test.go

pkg/streamer/file.go

D3Hunter · 2021-10-19T03:07:35Z

@Ehco1996

but if upstream have no write for a long time, does this cause reader read and parse same binlog file again and again (every watcherInterval), and this may cause much IO waste

we parse from the last position, so parseFile in go-mysql returns EOF if there nothing to read, for the FORMAT_DESC_EVENT(go-mysql parse it every time), page cache of os will handle it.

it would be better if we can keep the written size of each binlog file, so we'll know whether it's ended, maybe we can add it later.

lichunzhu

This control path looks a bit long to me.
Relay receives a new event -> Subtask holder -> Relay reader
The way I expect it to work is that notifications are only coordinated within the relay module, and the subtask holder only needs to request a reader from the relay.
Is this PR a transitional design or a final solution?

relay/relay.go

lichunzhu · 2021-10-19T04:53:18Z

dm/worker/subtask.go

@@ -44,12 +46,22 @@ const (
 	waitRelayCatchupTimeout = 30 * time.Second
 )

+type relayNotifier struct {
+	// ch with size = 1, we only need to be notified whether binlog file of relay changed, not how many times
+	ch chan interface{}


Suggested change

ch chan interface{}

ch chan struct{}

It has some cost to convert struct{}{} to interface{}

pkg/streamer/file.go

D3Hunter · 2021-10-19T08:19:18Z

@lichunzhu

The way I expect it to work is that notifications are only coordinated within the relay module, and the subtask holder only needs to request a reader from the relay.

i think it should be separate into relay and local binlog reader as current implementation do, that's 2 different things

relay responsible for reading from upstream and write into local file, and notify those are interested in event changes.
local binlog reader read binlog locally

Is this PR a transitional design or a final solution?

this PR is for optimizing latency with relay on, don't want to introduce too much refactoring

lance6716 · 2021-10-25T12:01:34Z

[code=38032:class=dm-master:scope=internal:level=high], Message: some error occurs in dm-worker: ErrCode:11106 ErrClass:\"functional\" ErrScope:\"internal\" ErrLevel:\"high\" Message:\"fail to restart streamer controller: the given relay log pos (dm-integration-test-s5hcg-7zsh8-bin.000002, 11232) of meta config is too big, please check it again\" Workaround:\"If the size of the corresponding binlog file has exceeded 4GB, please follow the solution in https://docs.pingcap.com/tidb-data-migration/stable/error-handling#the-relay-unit-throws-error-event-from--in--diff-from-passed-in-event--or-a-replication-task-is-interrupted-with-failing-to-get-or-parse-binlog-errors-like-get-binlog-error-error-1236-hy000-and-binlog-checksum-mismatch-data-may-be-corrupted-returned\" , Workaround: Please execute `query-status` to check status.

seems a BUG revealed in CI

D3Hunter · 2021-10-26T05:51:47Z

[code=38032:class=dm-master:scope=internal:level=high], Message: some error occurs in dm-worker: ErrCode:11106 ErrClass:\"functional\" ErrScope:\"internal\" ErrLevel:\"high\" Message:\"fail to restart streamer controller: the given relay log pos (dm-integration-test-s5hcg-7zsh8-bin.000002, 11232) of meta config is too big, please check it again\" Workaround:\"If the size of the corresponding binlog file has exceeded 4GB, please follow the solution in https://docs.pingcap.com/tidb-data-migration/stable/error-handling#the-relay-unit-throws-error-event-from--in--diff-from-passed-in-event--or-a-replication-task-is-interrupted-with-failing-to-get-or-parse-binlog-errors-like-get-binlog-error-error-1236-hy000-and-binlog-checksum-mismatch-data-may-be-corrupted-returned\" , Workaround: Please execute `query-status` to check status.

seems a BUG revealed in CI

it's an unstable case, not a bug. it happens when we

enable-relay, syncer'll restart and checks whether binlog file size < current sync pos, but in this case relay hasn't catch up with current sync pos due to higher load on CI, so syncer reports an error and pause.
then we try to stop the task, and this case fail. actually, syncer'll try again later and success, so it can be stopped.

it's hard to reproduce this locally(add sleep(30 ms) after each relay write do reproduce it.)

Ehco1996 · 2021-10-26T05:58:10Z

/run-unit-tests
/run-integration-test

Ehco1996 · 2021-10-26T06:00:05Z

but in this case relay hasn't catch up with current sync pos

maybe we can add some msg in workaround such as wait and retry 😂

dm/worker/server_test.go

dm/worker/source_worker_test.go

lichunzhu · 2021-10-26T09:28:26Z

dm/worker/subtask.go

@@ -44,12 +46,22 @@ const (
 	waitRelayCatchupTimeout = 30 * time.Second
 )

+type relayNotifier struct {
+	// ch with size = 1, we only need to be notified whether binlog file of relay changed, not how many times
+	ch chan interface{}


It has some cost to convert struct{}{} to interface{}

dm/worker/subtask.go

pkg/streamer/file_test.go

pkg/streamer/reader_test.go

lichunzhu · 2021-10-26T11:37:55Z

dm/worker/subtask_holder.go

+// OnEvent implements relay.Listener
+// only syncer unit of subtask need to be notified, but it's much simpler and less error-prone to manage it here
+// as relay event need to broadcast to every syncer(most subtask have a syncer).
+func (h *subTaskHolder) OnEvent(e *replication.BinlogEvent) {


Suggested change

func (h *subTaskHolder) OnEvent(e *replication.BinlogEvent) {

func (h *subTaskHolder) OnEvent(_ *replication.BinlogEvent) {

relay/relay.go

dm/worker/subtask_test.go

dm/worker/subtask.go

syncer/syncer.go

lichunzhu

Rest LGTM except for the unresolved comments

relay/relay_test.go

pkg/streamer/file_test.go

Ehco1996 · 2021-10-27T06:53:50Z

/merge

ti-chi-bot · 2021-10-27T06:53:53Z

This pull request has been accepted and is ready to merge.

Commit hash: 99b2ed7

…cy (pingcap#2225)

ti-chi-bot · 2021-10-27T06:55:28Z

In response to a cherrypick label: new pull request created: #2269.

D3Hunter added 7 commits October 15, 2021 11:52

notify syncer on new binlog event

838d439

reader notify streamer reader

7a14bef

move switch path check into binlog file change check

e4f2c7a

add timer to relayLogUpdatedOrNewCreated to handle special case

b067a07

ignore duplicate FORMAT event

41568ea

add/fix unit test

29c9fd4

make lint happy

9df7eed

ti-chi-bot added the contribution For contributor label Oct 15, 2021

ti-chi-bot requested review from GMHDBJD and lance6716 October 15, 2021 04:11

ti-chi-bot added the size/XXL label Oct 15, 2021

D3Hunter added 2 commits October 15, 2021 12:15

add comments to public if

4b2aa47

fix comments

b865044

ti-chi-bot requested a review from lichunzhu October 15, 2021 04:23

lance6716 added needs-cherry-pick-release-2.0 This PR should be cherry-picked to release-2.0. Remove this label after cherry-picked to release-2.0 needs-update-release-note This PR should be added into release notes. Remove this label once the release notes are updated labels Oct 15, 2021

Ehco1996 reviewed Oct 18, 2021

View reviewed changes

lance6716 reviewed Oct 18, 2021

View reviewed changes

relay/relay.go Show resolved Hide resolved

lance6716 reviewed Oct 18, 2021

View reviewed changes

D3Hunter added 5 commits October 18, 2021 16:24

fix lint

3ab6475

abstruct notifier chan in syncer into a struct and pass in from subtask

ea8fd6e

rename fields to make code more verbose

64d388f

add comment

ac2f391

add comments

af5f7e2

Ehco1996 reviewed Oct 19, 2021

View reviewed changes

lichunzhu reviewed Oct 19, 2021

View reviewed changes

Ehco1996 added this to the v2.1.0 milestone Oct 26, 2021

ti-chi-bot added the needs-rebase label Oct 26, 2021

D3Hunter added 3 commits October 26, 2021 14:29

fix unstable integration test case ha_cases3

026d9fb

Merge remote-tracking branch 'upstream/master' into relay-notify

2213223

fix comments

11511a3

ti-chi-bot removed the needs-rebase label Oct 26, 2021

undo redirect stderr in run_sql, some case depends on the output

9e93be1

lichunzhu reviewed Oct 26, 2021

View reviewed changes

dm/worker/subtask_test.go Outdated Show resolved Hide resolved

dm/worker/subtask.go Outdated Show resolved Hide resolved

syncer/syncer.go Show resolved Hide resolved

D3Hunter added 2 commits October 27, 2021 10:52

add more unit test

ebd1d2b

fix import groups

99b2ed7

lichunzhu reviewed Oct 27, 2021

View reviewed changes

relay/relay_test.go Show resolved Hide resolved

pkg/streamer/file_test.go Show resolved Hide resolved

lichunzhu approved these changes Oct 27, 2021

View reviewed changes

ti-chi-bot added the status/LGT1 One reviewer already commented LGTM label Oct 27, 2021

Ehco1996 approved these changes Oct 27, 2021

View reviewed changes

ti-chi-bot added status/LGT2 Two reviewers already commented LGTM, ready for merge and removed status/LGT1 One reviewer already commented LGTM labels Oct 27, 2021

ti-chi-bot added the status/can-merge label Oct 27, 2021

ti-chi-bot merged commit 8aa3734 into pingcap:master Oct 27, 2021

ti-chi-bot pushed a commit to ti-chi-bot/dm that referenced this pull request Oct 27, 2021

relay/syncer: relay notifies syncer of new write to reduce sync laten…

151cc17

…cy (pingcap#2225)

ti-chi-bot mentioned this pull request Oct 27, 2021

relay/syncer: relay notifies syncer of new write to reduce sync latency (#2225) #2269

Closed

D3Hunter deleted the relay-notify branch October 27, 2021 07:27

D3Hunter mentioned this pull request Nov 10, 2021

dm-test: fix unstable case pingcap/tiflow#3365

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

relay/syncer: relay notifies syncer of new write to reduce sync latency #2225

relay/syncer: relay notifies syncer of new write to reduce sync latency #2225

D3Hunter commented Oct 15, 2021 •

edited

Loading

ti-chi-bot commented Oct 15, 2021 •

edited

Loading

D3Hunter commented Oct 15, 2021

Ehco1996 Oct 19, 2021

D3Hunter Oct 19, 2021

lichunzhu Oct 26, 2021

D3Hunter commented Oct 19, 2021

lichunzhu left a comment

lichunzhu Oct 19, 2021

D3Hunter Oct 19, 2021

lichunzhu Oct 26, 2021

D3Hunter commented Oct 19, 2021 •

edited

Loading

lance6716 commented Oct 25, 2021

D3Hunter commented Oct 26, 2021 •

edited

Loading

Ehco1996 commented Oct 26, 2021

Ehco1996 commented Oct 26, 2021

lichunzhu Oct 26, 2021

lichunzhu Oct 26, 2021

lichunzhu left a comment •

edited

Loading

Ehco1996 commented Oct 27, 2021

ti-chi-bot commented Oct 27, 2021

ti-chi-bot commented Oct 27, 2021

	func (h subTaskHolder) OnEvent(e replication.BinlogEvent) {
	func (h subTaskHolder) OnEvent(_ replication.BinlogEvent) {

relay/syncer: relay notifies syncer of new write to reduce sync latency #2225

relay/syncer: relay notifies syncer of new write to reduce sync latency #2225

Conversation

D3Hunter commented Oct 15, 2021 • edited Loading

What problem does this PR solve?

What is changed and how it works?

Check List

ti-chi-bot commented Oct 15, 2021 • edited Loading

D3Hunter commented Oct 15, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

D3Hunter commented Oct 19, 2021

lichunzhu 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

D3Hunter commented Oct 19, 2021 • edited Loading

lance6716 commented Oct 25, 2021

D3Hunter commented Oct 26, 2021 • edited Loading

Ehco1996 commented Oct 26, 2021

Ehco1996 commented Oct 26, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lichunzhu left a comment • edited Loading

Choose a reason for hiding this comment

Ehco1996 commented Oct 27, 2021

ti-chi-bot commented Oct 27, 2021

ti-chi-bot commented Oct 27, 2021

D3Hunter commented Oct 15, 2021 •

edited

Loading

ti-chi-bot commented Oct 15, 2021 •

edited

Loading

D3Hunter commented Oct 19, 2021 •

edited

Loading

D3Hunter commented Oct 26, 2021 •

edited

Loading

lichunzhu left a comment •

edited

Loading