prefer atomic values to atomic operations

not sure why there's no atomic.NewUint32,
so leave baseConnID alone

also update some random code with changes since 1.20 is now required
don't use mrand.Intn directly to avoid contention on global rng

Author: serprex

canal/canal.go

@@ -51,7 +51,7 @@ type Canal struct {
 	includeTableRegex []*regexp.Regexp
 	excludeTableRegex []*regexp.Regexp
 
-	delay *uint32
+	delay atomic.Uint32
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -85,8 +85,6 @@ func NewCanal(cfg *Config) (*Canal, error) {
 	}
 	c.master = &masterInfo{logger: c.cfg.Logger}
 
-	c.delay = new(uint32)
-
 	var err error
 
 	if err = c.prepareDumper(); err != nil {
@@ -195,7 +193,7 @@ func (c *Canal) prepareDumper() error {
 }
 
 func (c *Canal) GetDelay() uint32 {
-	return atomic.LoadUint32(c.delay)
+	return c.delay.Load()
 }
 
 // Run will first try to dump all data from MySQL master `mysqldump`,

canal/sync.go

@@ -2,7 +2,6 @@ package canal
 
 import (
 	"log/slog"
-	"sync/atomic"
 	"time"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
@@ -259,7 +258,7 @@ func (c *Canal) updateReplicationDelay(ev *replication.BinlogEvent) {
 	if now >= ev.Header.Timestamp {
 		newDelay = now - ev.Header.Timestamp
 	}
-	atomic.StoreUint32(c.delay, newDelay)
+	c.delay.Store(newDelay)
 }
 
 func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {

mysql/util.go

@@ -16,7 +16,6 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
-	"time"
 
 	"filippo.io/edwards25519"
 	"github.com/go-mysql-org/go-mysql/utils"
@@ -171,13 +170,10 @@ func AppendLengthEncodedInteger(b []byte, n uint64) []byte {
 
 func RandomBuf(size int) []byte {
 	buf := make([]byte, size)
-	// When this project supports golang 1.20 as a minimum, then this mrand.New(...)
-	// line can be eliminated and the random number can be generated by simply
-	// calling mrand.Intn()
-	random := mrand.New(mrand.NewSource(time.Now().UTC().UnixNano()))
-	min, max := 30, 127
+	random := mrand.New(mrand.NewSource(int64(mrand.Uint64())))
+	min, max := int32(30), int32(127)
 	for i := 0; i < size; i++ {
-		buf[i] = byte(min + random.Intn(max-min))
+		buf[i] = byte(min + random.Int31n(max-min))
 	}
 	return buf
 }

replication/parser.go

@@ -33,7 +33,7 @@ type BinlogParser struct {
 	timestampStringLocation *time.Location
 
 	// used to start/stop processing
-	stopProcessing uint32
+	stopProcessing atomic.Bool
 
 	useDecimal               bool
 	useFloatWithTrailingZero bool
@@ -54,11 +54,11 @@ func NewBinlogParser() *BinlogParser {
 }
 
 func (p *BinlogParser) Stop() {
-	atomic.StoreUint32(&p.stopProcessing, 1)
+	p.stopProcessing.Store(true)
 }
 
 func (p *BinlogParser) Resume() {
-	atomic.StoreUint32(&p.stopProcessing, 0)
+	p.stopProcessing.Store(false)
 }
 
 func (p *BinlogParser) Reset() {
@@ -166,7 +166,7 @@ func (p *BinlogParser) parseSingleEvent(r io.Reader, onEvent OnEventFunc) (bool,
 }
 
 func (p *BinlogParser) ParseReader(r io.Reader, onEvent OnEventFunc) error {
-	for atomic.LoadUint32(&p.stopProcessing) != 1 {
+	for !p.stopProcessing.Load() {
 		done, err := p.parseSingleEvent(r, onEvent)
 		if err != nil {
 			if err == errMissingTableMapEvent {