Skip to content
This repository has been archived by the owner on Nov 24, 2023. It is now read-only.

poor: resume task if sync unit exits with invalid connection error #66

Merged
merged 5 commits into from
Mar 7, 2019
Merged

poor: resume task if sync unit exits with invalid connection error #66

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
103b912
*: add poor retry
IANTHEREAL Mar 5, 2019
7eb04ce
refine code
IANTHEREAL Mar 5, 2019
12bd313
*: correct code
IANTHEREAL Mar 5, 2019
c6ae341
*: leave a todo comment
IANTHEREAL Mar 5, 2019
4754c9c
Update dm/worker/subtask.go
amyangfei Mar 7, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 42 additions & 8 deletions dm/worker/subtask.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
package worker

import (
"strings"
"sync"
"time"

Expand Down Expand Up @@ -154,7 +155,7 @@ func (st *SubTask) Run() {
st.ctx, st.cancel = context.WithCancel(context.Background())
pr := make(chan pb.ProcessResult, 1)
st.wg.Add(1)
go st.fetchResult(st.ctx, st.cancel, pr)
go st.fetchResult(pr)
go cu.Process(st.ctx, pr)

st.wg.Add(1)
Expand All @@ -163,34 +164,50 @@ func (st *SubTask) Run() {

// fetchResult fetches units process result
// when dm-unit report an error, we need to re-Process the sub task
func (st *SubTask) fetchResult(ctx context.Context, cancel context.CancelFunc, pr chan pb.ProcessResult) {
func (st *SubTask) fetchResult(pr chan pb.ProcessResult) {
defer st.wg.Done()

retry:
select {
case <-ctx.Done():
case <-st.ctx.Done():
return
case result := <-pr:
st.setResult(&result) // save result
cancel() // dm-unit finished, canceled or error occurred, always cancel processing
st.cancel() // dm-unit finished, canceled or error occurred, always cancel processing

if len(result.Errors) == 0 && st.Stage() == pb.Stage_Paused {
return // paused by external request
}

var stage pb.Stage
var (
cu = st.CurrUnit()
stage pb.Stage
)
if len(result.Errors) == 0 {
if result.IsCanceled {
stage = pb.Stage_Stopped // canceled by user
} else {
stage = pb.Stage_Finished // process finished with no error
}
} else {
/* TODO
it's a poor and very rough retry feature, the main reason is that
the concurrency control of the sub task module is very confusing and needs to be optimized.
Copy link
Contributor

@amyangfei amyangfei Mar 6, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the concurrency control optimization including the following scenario?
all of DM-worker's connections to downstream TiDB suddenly reset by downstream(caused by TiDB restart, network cutting etc.), then we must recover each connection in syncer and if the worker-count config of syncer is N, we will wait (N + 1) * 10 seconds at most.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

first of all it is concurrent, I don't understand why we need to wait (N + 1) * 10 seconds at most.
For the refactoring, more is the reconstruction of the the structure, making the moudle and funcion more reasone, of course we can increase the backoff property

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean for each subtask, (N + 1) * 10 seconds may be not continuous, but each DB connection must be recovered. So the worst case is (N + 1) * 10 seconds sums up.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

but it's real concurrent, right?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's the logic of unit, we can refine while refactoring sync unit

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we can refactor units one by one and then the outer state transition later.

After improving its state transition and concurrency control,
I will optimize the implementation of retry feature.
*/
if st.retryErrors(result.Errors, cu) {
log.Warnf("[subtask] %s (%s) retry on error %v, waiting 10 seconds!", st.cfg.Name, cu.Type(), result.Errors)
st.ctx, st.cancel = context.WithCancel(context.Background())
time.Sleep(10 * time.Second)
go cu.Resume(st.ctx, pr)
goto retry
}

stage = pb.Stage_Paused // error occurred, paused
}
st.setStage(stage)

cu := st.CurrUnit()

log.Infof("[subtask] %s dm-unit %s process returned with stage %s, status %s", st.cfg.Name, cu.Type(), stage.String(), st.StatusJSON())

switch stage {
Expand Down Expand Up @@ -391,7 +408,7 @@ func (st *SubTask) Resume() error {
st.ctx, st.cancel = context.WithCancel(context.Background())
pr := make(chan pb.ProcessResult, 1)
st.wg.Add(1)
go st.fetchResult(st.ctx, st.cancel, pr)
go st.fetchResult(pr)
go cu.Resume(st.ctx, pr)

st.wg.Add(1)
Expand Down Expand Up @@ -621,3 +638,20 @@ func (st *SubTask) unitTransWaitCondition() error {
}
return nil
}

func (st *SubTask) retryErrors(errors []*pb.ProcessError, current unit.Unit) bool {
retry := true
switch current.Type() {
case pb.UnitType_Sync:
for _, err := range errors {
if strings.Contains(err.Msg, "invalid connection") {
Copy link
Member

@csuzhangxc csuzhangxc Mar 7, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we need to close all DB connections and re-open them in sync unit?
some users reported that they need to resume-task multi times when "invalid connection" occurred.

continue
}
retry = false
}
default:
retry = false
}

return retry
}