Replication delay should be updated with atomic.StoreUint32 (#392)

narkq · siddontang · commit 9b0760372c60 · 2019-06-15T21:21:31.000+08:00
diff --git a/canal/sync.go b/canal/sync.go
@@ -213,7 +213,12 @@ func (c *Canal) updateTable(db, table string) (err error) {
 	return
 }
 func (c *Canal) updateReplicationDelay(ev *replication.BinlogEvent) {
-	atomic.AddUint32(c.delay, uint32(time.Now().Unix())-ev.Header.Timestamp)
+	var newDelay uint32
+	now := uint32(time.Now().Unix())
+	if now >= ev.Header.Timestamp {
+		newDelay = now - ev.Header.Timestamp
+	}
+	atomic.StoreUint32(c.delay, newDelay)
 }
 
 func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {

Original file line number	Diff line number	Diff line change
`@@ -213,7 +213,12 @@ func (c *Canal) updateTable(db, table string) (err error) {`
`213`	`213`	`return`
`214`	`214`	`}`
`215`	`215`	`func (c Canal) updateReplicationDelay(ev replication.BinlogEvent) {`
`216`		`- atomic.AddUint32(c.delay, uint32(time.Now().Unix())-ev.Header.Timestamp)`
	`216`	`+ var newDelay uint32`
	`217`	`+ now := uint32(time.Now().Unix())`
	`218`	`+ if now >= ev.Header.Timestamp {`
	`219`	`+ newDelay = now - ev.Header.Timestamp`
	`220`	`+ }`
	`221`	`+ atomic.StoreUint32(c.delay, newDelay)`
`217`	`222`	`}`
`218`	`223`
`219`	`224`	`func (c Canal) handleRowsEvent(e replication.BinlogEvent) error {`