Reduce "busy buffer" logs

buffer.go

@@ -43,6 +43,11 @@ func newBuffer(nc net.Conn) buffer {
 	}
 }
 
+// busy returns true if the buffer contains some read data.
+func (b *buffer) busy() bool {
+	return b.length > 0
+}
+
 // flip replaces the active buffer with the background buffer
 // this is a delayed flip that simply increases the buffer counter;
 // the actual flip will be performed the next time we call `buffer.fill`

connection.go

@@ -121,10 +121,14 @@ func (mc *mysqlConn) Close() (err error) {
 	if !mc.closed.Load() {
 		err = mc.writeCommandPacket(comQuit)
 	}
+	mc.close()
+	return
+}
 
+// close closes the network connection and clear results without sending COM_QUIT.
+func (mc *mysqlConn) close() {
 	mc.cleanup()
 	mc.clearResult()
-	return
 }
 
 // Closes the network connection and unsets internal variables. Do not call this
@@ -637,7 +641,7 @@ func (mc *mysqlConn) CheckNamedValue(nv *driver.NamedValue) (err error) {
 // ResetSession implements driver.SessionResetter.
 // (From Go 1.10)
 func (mc *mysqlConn) ResetSession(ctx context.Context) error {
-	if mc.closed.Load() {
+	if mc.closed.Load() || mc.buf.busy() {
 		return driver.ErrBadConn
 	}
 
@@ -671,7 +675,7 @@ func (mc *mysqlConn) ResetSession(ctx context.Context) error {
 // IsValid implements driver.Validator interface
 // (From Go 1.15)
 func (mc *mysqlConn) IsValid() bool {
-	return !mc.closed.Load()
+	return !mc.closed.Load() && !mc.buf.busy()
 }
 
 var _ driver.SessionResetter = &mysqlConn{}

packets.go

@@ -32,11 +32,11 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 		// read packet header
 		data, err := mc.buf.readNext(4)
 		if err != nil {
+			mc.close()
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return nil, cerr
 			}
 			mc.log(err)
-			mc.Close()
 			return nil, ErrInvalidConn
 		}
 
@@ -45,7 +45,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 
 		// check packet sync [8 bit]
 		if data[3] != mc.sequence {
-			mc.Close()
+			mc.close()
 			if data[3] > mc.sequence {
 				return nil, ErrPktSyncMul
 			}
@@ -59,7 +59,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 			// there was no previous packet
 			if prevData == nil {
 				mc.log(ErrMalformPkt)
-				mc.Close()
+				mc.close()
 				return nil, ErrInvalidConn
 			}
 
@@ -69,11 +69,11 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 		// read packet body [pktLen bytes]
 		data, err = mc.buf.readNext(pktLen)
 		if err != nil {
+			mc.close()
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return nil, cerr
 			}
 			mc.log(err)
-			mc.Close()
 			return nil, ErrInvalidConn
 		}
 
@@ -125,10 +125,10 @@ func (mc *mysqlConn) writePacket(data []byte) error {
 
 		n, err := mc.netConn.Write(data[:4+size])
 		if err != nil {
+			mc.cleanup()
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return cerr
 			}
-			mc.cleanup()
 			if n == 0 && pktLen == len(data)-4 {
 				// only for the first loop iteration when nothing was written yet
 				mc.log(err)
@@ -162,11 +162,6 @@ func (mc *mysqlConn) writePacket(data []byte) error {
 func (mc *mysqlConn) readHandshakePacket() (data []byte, plugin string, err error) {
 	data, err = mc.readPacket()
 	if err != nil {
-		// for init we can rewrite this to ErrBadConn for sql.Driver to retry, since
-		// in connection initialization we don't risk retrying non-idempotent actions.
-		if err == ErrInvalidConn {
-			return nil, "", driver.ErrBadConn
-		}
 		return
 	}
 
@@ -312,9 +307,8 @@ func (mc *mysqlConn) writeHandshakeResponsePacket(authResp []byte, plugin string
 	// Calculate packet length and get buffer with that size
 	data, err := mc.buf.takeBuffer(pktLen + 4)
 	if err != nil {
-		// cannot take the buffer. Something must be wrong with the connection
-		mc.log(err)
-		return errBadConnNoWrite
+		mc.cleanup()
+		return err
 	}
 
 	// ClientFlags [32 bit]
@@ -404,9 +398,8 @@ func (mc *mysqlConn) writeAuthSwitchPacket(authData []byte) error {
 	pktLen := 4 + len(authData)
 	data, err := mc.buf.takeBuffer(pktLen)
 	if err != nil {
-		// cannot take the buffer. Something must be wrong with the connection
-		mc.log(err)
-		return errBadConnNoWrite
+		mc.cleanup()
+		return err
 	}
 
 	// Add the auth data [EOF]
@@ -424,9 +417,7 @@ func (mc *mysqlConn) writeCommandPacket(command byte) error {
 
 	data, err := mc.buf.takeSmallBuffer(4 + 1)
 	if err != nil {
-		// cannot take the buffer. Something must be wrong with the connection
-		mc.log(err)
-		return errBadConnNoWrite
+		return err
 	}
 
 	// Add command byte
@@ -443,9 +434,7 @@ func (mc *mysqlConn) writeCommandPacketStr(command byte, arg string) error {
 	pktLen := 1 + len(arg)
 	data, err := mc.buf.takeBuffer(pktLen + 4)
 	if err != nil {
-		// cannot take the buffer. Something must be wrong with the connection
-		mc.log(err)
-		return errBadConnNoWrite
+		return err
 	}
 
 	// Add command byte
@@ -464,9 +453,7 @@ func (mc *mysqlConn) writeCommandPacketUint32(command byte, arg uint32) error {
 
 	data, err := mc.buf.takeSmallBuffer(4 + 1 + 4)
 	if err != nil {
-		// cannot take the buffer. Something must be wrong with the connection
-		mc.log(err)
-		return errBadConnNoWrite
+		return err
 	}
 
 	// Add command byte
@@ -1007,9 +994,7 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 		// In this case the len(data) == cap(data) which is used to optimise the flow below.
 	}
 	if err != nil {
-		// cannot take the buffer. Something must be wrong with the connection
-		mc.log(err)
-		return errBadConnNoWrite
+		return err
 	}
 
 	// command [1 byte]
@@ -1207,8 +1192,7 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 		if valuesCap != cap(paramValues) {
 			data = append(data[:pos], paramValues...)
 			if err = mc.buf.store(data); err != nil {
-				mc.log(err)
-				return errBadConnNoWrite
+				return err
 			}
 		}