relay: use Reader interface read binlog events #92
Conversation
csuzhangxc
added
priority/normal
|
/run-all-tests |
csuzhangxc
changed the title
csuzhangxc
added
status/PTAL
|
@GregoryIan @amyangfei PTAL |
pkg/binlog/reader/tcp.go
Outdated
| // TCPReaderStatus represents the status of a TCPReader. | ||
| type TCPReaderStatus struct { | ||
| Stage string `json:"stage"` | ||
| Connection uint32 `json:"connection"` |
There was a problem hiding this comment.
what's meaning of Connection?
There was a problem hiding this comment.
renamed to ConnID, but keep the json flag as connection.
pkg/binlog/reader/tcp.go
Outdated
| defer r.stageMu.Unlock() | ||
|
|
||
| if r.stage != stageNew { | ||
| return errors.NotValidf("stage %d, expect %d", r.stage, stageNew) |
There was a problem hiding this comment.
is log stage %d, expect %d is not valid?
There was a problem hiding this comment.
changed to stage %s, expect %s only.
pkg/binlog/reader/tcp.go
Outdated
| r.syncerCfg.User, r.syncerCfg.Password, r.syncerCfg.Host, r.syncerCfg.Port) | ||
| db, err := sql.Open("mysql", dsn) | ||
| if err != nil { | ||
| return errors.Annotate(err, "open connection to the master") |
There was a problem hiding this comment.
I think we can output which master
pkg/binlog/reader/tcp.go
Outdated
| defer db.Close() | ||
| err = utils.KillConn(db, connID) | ||
| if err != nil { | ||
| return errors.Annotatef(err, "kill connection %d", connID) |
pkg/binlog/reader/tcp.go
Outdated
| func (r *TCPReader) GetEvent(ctx context.Context, checkStage bool) (*replication.BinlogEvent, error) { | ||
| if checkStage { | ||
| r.stageMu.Lock() | ||
| if r.stage != stagePrepared { |
There was a problem hiding this comment.
would atomic operation be better?
There was a problem hiding this comment.
changed to use atomic operation, and removed checkStage.
pkg/binlog/reader/tcp.go
Outdated
|
|
||
| // StartSyncByGTID implements Reader.StartSyncByGTID. | ||
| func (r *TCPReader) StartSyncByGTID(gSet gtid.Set) error { | ||
| r.stageMu.Lock() |
There was a problem hiding this comment.
in fact this lock prevent more than stage field
|
Rest LGTM. well done |
|
/run-all-tests |
pkg/binlog/reader/tcp.go
Outdated
| func (r *TCPReader) Status() interface{} { | ||
| r.mu.Lock() | ||
| stage := r.stage | ||
| r.mu.Unlock() |
There was a problem hiding this comment.
why only lock for stage getter? besides I find some redundant lock protection, such as
r.mu.Lock()
defer r.mu.Unlock()
r.stage.Get()
Is there any consideration to do that?
There was a problem hiding this comment.
I change to use stage = r.stage.Get without lock.
We use stage both in comparing(L150) and recording(L154).
Saving stage in L148, so we can get the same value both in L150 and L154.
|
LGTM |
IANTHEREAL
added
status/LGT1
csuzhangxc
added
the
status/LGT2
What problem does this PR solve?
This PR is a part work of #91
Introduced a binlog event
Readerinterface, so we can implement at least 3 kinds of Reader later:In this PR, I only implement a
TCPReaderand refactor a litter code for relay writer to make review easier.Later, the following things will be done:
What is changed and how it works?
TCPReaderTCPReaderCheck List
Tests
Code changes