Opt: change timeout to atomic

connection.go

@@ -82,10 +82,10 @@ type gettyConn struct {
 	readPkgNum    uatomic.Uint32 // send pkg number
 	writePkgNum   uatomic.Uint32 // recv pkg number
 	active        uatomic.Int64  // last active, in milliseconds
-	rTimeout      time.Duration  // network current limiting
-	wTimeout      time.Duration
-	rLastDeadline time.Time // last network read time
-	wLastDeadline time.Time // last network write time
+	rTimeout      uatomic.Duration // network current limiting
+	wTimeout      uatomic.Duration
+	rLastDeadline uatomic.Time // last network read time
+	wLastDeadline uatomic.Time // last network write time
 	local         string    // local address
 	peer          string    // peer address
 	ss            Session
@@ -126,7 +126,7 @@ func (c *gettyConn) send(interface{}) (int, error) {
 func (c *gettyConn) close(int) {}
 
 func (c gettyConn) readTimeout() time.Duration {
-	return c.rTimeout
+	return c.rTimeout.Load()
 }
 
 func (c *gettyConn) setSession(ss Session) {
@@ -142,14 +142,14 @@ func (c *gettyConn) SetReadTimeout(rTimeout time.Duration) {
 		panic("@rTimeout < 1")
 	}
 
-	c.rTimeout = rTimeout
-	if c.wTimeout == 0 {
-		c.wTimeout = rTimeout
+	c.rTimeout.Store(rTimeout)
+	if c.wTimeout.Load() == 0 {
+		c.wTimeout.Store(rTimeout)
 	}
 }
 
 func (c gettyConn) writeTimeout() time.Duration {
-	return c.wTimeout
+	return c.wTimeout.Load()
 }
 
 // SetWriteTimeout Pls do not set write deadline for websocket connection. AlexStocks 20180310
@@ -161,9 +161,9 @@ func (c *gettyConn) SetWriteTimeout(wTimeout time.Duration) {
 		panic("@wTimeout < 1")
 	}
 
-	c.wTimeout = wTimeout
-	if c.rTimeout == 0 {
-		c.rTimeout = wTimeout
+	c.wTimeout.Store(wTimeout)
+	if c.rTimeout.Load() == 0 {
+		c.rTimeout.Store(wTimeout)
 	}
 }
 
@@ -198,8 +198,8 @@ func newGettyTCPConn(conn net.Conn) *gettyTCPConn {
 		writer: io.Writer(conn),
 		gettyConn: gettyConn{
 			id:       connID.Add(1),
-			rTimeout: netIOTimeout,
-			wTimeout: netIOTimeout,
+			rTimeout: *uatomic.NewDuration(netIOTimeout),
+			wTimeout: *uatomic.NewDuration(netIOTimeout),
 			local:    localAddr,
 			peer:     peerAddr,
 			compress: CompressNone,
@@ -266,17 +266,17 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 	)
 
 	// set read timeout deadline
-	if t.compress == CompressNone && t.rTimeout > 0 {
+	if t.compress == CompressNone && t.rTimeout.Load() > 0 {
 		// Optimization: update read deadline only if more than 25%
 		// of the last read deadline exceeded.
 		// See https://github.com/golang/go/issues/15133 for details.
 		currentTime = time.Now()
-		if currentTime.Sub(t.rLastDeadline) > t.rTimeout>>2 {
-			if err = t.conn.SetReadDeadline(currentTime.Add(t.rTimeout)); err != nil {
+		if currentTime.Sub(t.rLastDeadline.Load()) > t.rTimeout.Load()>>2 {
+			if err = t.conn.SetReadDeadline(currentTime.Add(t.rTimeout.Load())); err != nil {
 				// just a timeout error
 				return 0, perrors.WithStack(err)
 			}
-			t.rLastDeadline = currentTime
+			t.rLastDeadline.Store(currentTime)
 		}
 	}
 
@@ -296,16 +296,16 @@ func (t *gettyTCPConn) send(pkg interface{}) (int, error) {
 		lg          int64
 	)
 
-	if t.compress == CompressNone && t.wTimeout > 0 {
+	if t.compress == CompressNone && t.wTimeout.Load() > 0 {
 		// Optimization: update write deadline only if more than 25%
 		// of the last write deadline exceeded.
 		// See https://github.com/golang/go/issues/15133 for details.
 		currentTime = time.Now()
-		if currentTime.Sub(t.wLastDeadline) > t.wTimeout>>2 {
-			if err = t.conn.SetWriteDeadline(currentTime.Add(t.wTimeout)); err != nil {
+		if currentTime.Sub(t.wLastDeadline.Load()) > t.wTimeout.Load()>>2 {
+			if err = t.conn.SetWriteDeadline(currentTime.Add(t.wTimeout.Load())); err != nil {
 				return 0, perrors.WithStack(err)
 			}
-			t.wLastDeadline = currentTime
+			t.wLastDeadline.Store(currentTime)
 		}
 	}
 
@@ -396,8 +396,8 @@ func newGettyUDPConn(conn *net.UDPConn) *gettyUDPConn {
 		conn: conn,
 		gettyConn: gettyConn{
 			id:       connID.Add(1),
-			rTimeout: netIOTimeout,
-			wTimeout: netIOTimeout,
+			rTimeout: *uatomic.NewDuration(netIOTimeout),
+			wTimeout: *uatomic.NewDuration(netIOTimeout),
 			local:    localAddr,
 			peer:     peerAddr,
 			compress: CompressNone,
@@ -417,16 +417,16 @@ func (u *gettyUDPConn) SetCompressType(c CompressType) {
 
 // udp connection read
 func (u *gettyUDPConn) recv(p []byte) (int, *net.UDPAddr, error) {
-	if u.rTimeout > 0 {
+	if u.rTimeout.Load() > 0 {
 		// Optimization: update read deadline only if more than 25%
 		// of the last read deadline exceeded.
 		// See https://github.com/golang/go/issues/15133 for details.
 		currentTime := time.Now()
-		if currentTime.Sub(u.rLastDeadline) > u.rTimeout>>2 {
-			if err := u.conn.SetReadDeadline(currentTime.Add(u.rTimeout)); err != nil {
+		if currentTime.Sub(u.rLastDeadline.Load()) > u.rTimeout.Load()>>2 {
+			if err := u.conn.SetReadDeadline(currentTime.Add(u.rTimeout.Load())); err != nil {
 				return 0, nil, perrors.WithStack(err)
 			}
-			u.rLastDeadline = currentTime
+			u.rLastDeadline.Store(currentTime)
 		}
 	}
 
@@ -464,16 +464,16 @@ func (u *gettyUDPConn) send(udpCtx interface{}) (int, error) {
 		}
 	}
 
-	if u.wTimeout > 0 {
+	if u.wTimeout.Load() > 0 {
 		// Optimization: update write deadline only if more than 25%
 		// of the last write deadline exceeded.
 		// See https://github.com/golang/go/issues/15133 for details.
 		currentTime = time.Now()
-		if currentTime.Sub(u.wLastDeadline) > u.wTimeout>>2 {
-			if err = u.conn.SetWriteDeadline(currentTime.Add(u.wTimeout)); err != nil {
+		if currentTime.Sub(u.wLastDeadline.Load()) > u.wTimeout.Load()>>2 {
+			if err = u.conn.SetWriteDeadline(currentTime.Add(u.wTimeout.Load())); err != nil {
 				return 0, perrors.WithStack(err)
 			}
-			u.wLastDeadline = currentTime
+			u.wLastDeadline.Store(currentTime)
 		}
 	}
 
@@ -521,8 +521,8 @@ func newGettyWSConn(conn *websocket.Conn) *gettyWSConn {
 		conn: conn,
 		gettyConn: gettyConn{
 			id:       connID.Add(1),
-			rTimeout: netIOTimeout,
-			wTimeout: netIOTimeout,
+			rTimeout: *uatomic.NewDuration(netIOTimeout),
+			wTimeout: *uatomic.NewDuration(netIOTimeout),
 			local:    localAddr,
 			peer:     peerAddr,
 			compress: CompressNone,
@@ -589,16 +589,16 @@ func (w *gettyWSConn) updateWriteDeadline() error {
 		currentTime time.Time
 	)
 
-	if w.wTimeout > 0 {
+	if w.wTimeout.Load() > 0 {
 		// Optimization: update write deadline only if more than 25%
 		// of the last write deadline exceeded.
 		// See https://github.com/golang/go/issues/15133 for details.
 		currentTime = time.Now()
-		if currentTime.Sub(w.wLastDeadline) > w.wTimeout>>2 {
-			if err = w.conn.SetWriteDeadline(currentTime.Add(w.wTimeout)); err != nil {
+		if currentTime.Sub(w.wLastDeadline.Load()) > w.wTimeout.Load()>>2 {
+			if err = w.conn.SetWriteDeadline(currentTime.Add(w.wTimeout.Load())); err != nil {
 				return perrors.WithStack(err)
 			}
-			w.wLastDeadline = currentTime
+			w.wLastDeadline.Store(currentTime)
 		}
 	}
 

go.mod

@@ -9,6 +9,6 @@ require (
 	github.com/montanaflynn/stats v0.6.6
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.7.0
-	go.uber.org/atomic v1.7.0
+	go.uber.org/atomic v1.9.0
 	go.uber.org/zap v1.16.0
 )

go.sum

@@ -413,8 +413,9 @@ go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
-go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
+go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
 go.uber.org/multierr v1.5.0 h1:KCa4XfM8CWFCpxXRGok+Q0SS/0XBhMDbHHGABQLvD2A=