Move to Go 1.19 atomics

go.mod

@@ -113,6 +113,7 @@ require (
 	github.com/kr/text v0.2.0
 	github.com/nsf/jsondiff v0.0.0-20210926074059-1e845ec5d249
 	golang.org/x/exp v0.0.0-20230131160201-f062dba9d201
+	golang.org/x/sync v0.1.0
 	modernc.org/sqlite v1.20.3
 )
 

go/mysql/conn.go

@@ -26,6 +26,7 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"vitess.io/vitess/go/mysql/collations"
@@ -34,7 +35,6 @@ import (
 
 	"vitess.io/vitess/go/bucketpool"
 	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/sync2"
 	"vitess.io/vitess/go/vt/log"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
@@ -161,7 +161,7 @@ type Conn struct {
 	Capabilities uint32
 
 	// closed is set to true when Close() is called on the connection.
-	closed sync2.AtomicBool
+	closed atomic.Bool
 
 	// ConnectionID is set:
 	// - at Connect() time for clients, with the value returned by
@@ -236,7 +236,6 @@ var readersPool = sync.Pool{New: func() any { return bufio.NewReaderSize(nil, co
 func newConn(conn net.Conn) *Conn {
 	return &Conn{
 		conn:           conn,
-		closed:         sync2.NewAtomicBool(false),
 		bufferedReader: bufio.NewReaderSize(conn, connBufferSize),
 	}
 }
@@ -250,7 +249,6 @@ func newServerConn(conn net.Conn, listener *Listener) *Conn {
 	c := &Conn{
 		conn:        conn,
 		listener:    listener,
-		closed:      sync2.NewAtomicBool(false),
 		PrepareData: make(map[uint32]*PrepareData),
 	}
 
@@ -717,7 +715,7 @@ func (c *Conn) Close() {
 // Close() method.  Note if the other side closes the connection, but
 // Close() wasn't called, this will return false.
 func (c *Conn) IsClosed() bool {
-	return c.closed.Get()
+	return c.closed.Load()
 }
 
 //
@@ -1294,7 +1292,7 @@ func (c *Conn) handleComSetOption(data []byte) bool {
 func (c *Conn) handleComPing() bool {
 	c.recycleReadPacket()
 	// Return error if listener was shut down and OK otherwise
-	if c.listener.isShutdown() {
+	if c.listener.shutdown.Load() {
 		if !c.writeErrorAndLog(ERServerShutdown, SSNetError, "Server shutdown in progress") {
 			return false
 		}

go/mysql/handshake_test.go

@@ -71,7 +71,7 @@ func TestClearTextClientAuth(t *testing.T) {
 	}
 
 	// Change server side to allow clear text without auth.
-	l.AllowClearTextWithoutTLS.Set(true)
+	l.AllowClearTextWithoutTLS.Store(true)
 	conn, err := Connect(ctx, params)
 	require.NoError(t, err, "unexpected connection error: %v", err)
 

go/mysql/server.go

@@ -35,7 +35,6 @@ import (
 	"vitess.io/vitess/go/netutil"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/stats"
-	"vitess.io/vitess/go/sync2"
 	"vitess.io/vitess/go/tb"
 	"vitess.io/vitess/go/vt/log"
 	querypb "vitess.io/vitess/go/vt/proto/query"
@@ -175,11 +174,11 @@ type Listener struct {
 	// AllowClearTextWithoutTLS needs to be set for the
 	// mysql_clear_password authentication method to be accepted
 	// by the server when TLS is not in use.
-	AllowClearTextWithoutTLS sync2.AtomicBool
+	AllowClearTextWithoutTLS atomic.Bool
 
 	// SlowConnectWarnThreshold if non-zero specifies an amount of time
 	// beyond which a warning is logged to identify the slow connection
-	SlowConnectWarnThreshold sync2.AtomicDuration
+	SlowConnectWarnThreshold atomic.Int64
 
 	// The following parameters are changed by the Accept routine.
 
@@ -198,7 +197,7 @@ type Listener struct {
 	connBufferPooling bool
 
 	// shutdown indicates that Shutdown method was called.
-	shutdown sync2.AtomicBool
+	shutdown atomic.Bool
 
 	// RequireSecureTransport configures the server to reject connections from insecure clients
 	RequireSecureTransport bool
@@ -454,7 +453,7 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 			return
 		}
 
-		if !l.AllowClearTextWithoutTLS.Get() && !c.TLSEnabled() && !negotiatedAuthMethod.AllowClearTextWithoutTLS() {
+		if !l.AllowClearTextWithoutTLS.Load() && !c.TLSEnabled() && !negotiatedAuthMethod.AllowClearTextWithoutTLS() {
 			c.writeErrorPacket(CRServerHandshakeErr, SSUnknownSQLState, "Cannot use clear text authentication over non-SSL connections.")
 			return
 		}
@@ -514,8 +513,8 @@ func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Ti
 	timings.Record(connectTimingKey, acceptTime)
 
 	// Log a warning if it took too long to connect
-	connectTime := time.Since(acceptTime)
-	if threshold := l.SlowConnectWarnThreshold.Get(); threshold != 0 && connectTime > threshold {
+	connectTime := time.Since(acceptTime).Nanoseconds()
+	if threshold := l.SlowConnectWarnThreshold.Load(); threshold != 0 && connectTime > threshold {
 		connSlow.Add(1)
 		log.Warningf("Slow connection from %s: %v", c, connectTime)
 	}
@@ -545,10 +544,6 @@ func (l *Listener) Shutdown() {
 	}
 }
 
-func (l *Listener) isShutdown() bool {
-	return l.shutdown.Get()
-}
-
 // writeHandshakeV10 writes the Initial Handshake Packet, server side.
 // It returns the salt data.
 func (c *Conn) writeHandshakeV10(serverVersion string, authServer AuthServer, enableTLS bool) ([]byte, error) {

go/mysql/server_flaky_test.go

@@ -530,7 +530,7 @@ func TestServer(t *testing.T) {
 	defer authServer.close()
 	l, err := NewListener("tcp", "127.0.0.1:", authServer, th, 0, 0, false, false)
 	require.NoError(t, err)
-	l.SlowConnectWarnThreshold.Set(time.Nanosecond * 1)
+	l.SlowConnectWarnThreshold.Store(time.Nanosecond.Nanoseconds())
 	defer l.Close()
 	go l.Accept()
 
@@ -630,7 +630,7 @@ func TestServerStats(t *testing.T) {
 	defer authServer.close()
 	l, err := NewListener("tcp", "127.0.0.1:", authServer, th, 0, 0, false, false)
 	require.NoError(t, err)
-	l.SlowConnectWarnThreshold.Set(time.Nanosecond * 1)
+	l.SlowConnectWarnThreshold.Store(time.Nanosecond.Nanoseconds())
 	defer l.Close()
 	go l.Accept()
 
@@ -675,7 +675,7 @@ func TestServerStats(t *testing.T) {
 	}
 
 	// Set the slow connect threshold to something high that we don't expect to trigger
-	l.SlowConnectWarnThreshold.Set(time.Second * 1)
+	l.SlowConnectWarnThreshold.Store(time.Second.Nanoseconds())
 
 	// Run a 'panic' command, other side should panic, recover and
 	// close the connection.
@@ -723,7 +723,7 @@ func TestClearTextServer(t *testing.T) {
 	// Run a 'select rows' command with results.  This should fail
 	// as clear text is not enabled by default on the client
 	// (except MariaDB).
-	l.AllowClearTextWithoutTLS.Set(true)
+	l.AllowClearTextWithoutTLS.Store(true)
 	sql := "select rows"
 	output, ok := runMysql(t, params, sql)
 	if ok {
@@ -741,7 +741,7 @@ func TestClearTextServer(t *testing.T) {
 	}
 
 	// Now enable clear text plugin in client, but server requires SSL.
-	l.AllowClearTextWithoutTLS.Set(false)
+	l.AllowClearTextWithoutTLS.Store(false)
 	if !isMariaDB {
 		sql = enableCleartextPluginPrefix + sql
 	}
@@ -750,7 +750,7 @@ func TestClearTextServer(t *testing.T) {
 	assert.Contains(t, output, "Cannot use clear text authentication over non-SSL connections", "Unexpected output for 'select rows': %v", output)
 
 	// Now enable clear text plugin, it should now work.
-	l.AllowClearTextWithoutTLS.Set(true)
+	l.AllowClearTextWithoutTLS.Store(true)
 	output, ok = runMysql(t, params, sql)
 	require.True(t, ok, "mysql failed: %v", output)
 
@@ -777,7 +777,7 @@ func TestDialogServer(t *testing.T) {
 	defer authServer.close()
 	l, err := NewListener("tcp", "127.0.0.1:", authServer, th, 0, 0, false, false)
 	require.NoError(t, err)
-	l.AllowClearTextWithoutTLS.Set(true)
+	l.AllowClearTextWithoutTLS.Store(true)
 	defer l.Close()
 	go l.Accept()