7194530928

Committed 13 Dec 2023 11:25AM UTC coverage: 82.145% (-0.5%) from 82.65%

Build # 7194530928

Build Type

push

github

Committed by

web-flow

Commit Message

Parallelize test (#1525)

* Refactor test cleanup in driver_test.go

* parallelize TestEmptyQuery and TestCRUD

* parallelize TestNumbersToAny

* parallelize TestInt

* parallelize TestFloat32

* parallelize TestFloat64

* parallelize TestFloat64Placeholder

* parallelize TestString

* parallelize TestRawBytes

* parallelize TestRawMessage

* parallelize TestValuer

* parallelize TestValuerWithValidation

* parallelize TestTimestampMicros

* parallelize TestNULL

* parallelize TestUint64

* parallelize TestLongData

* parallelize TestContextCancelExec

* parallelize TestPingContext

* parallelize TestContextCancelQuery

* parallelize TestContextCancelQueryRow

* Revert "parallelize TestLongData"

This reverts commit a360be7a1.

* parallelize TestContextCancelPrepare

* parallelize TestContextCancelStmtExec

* parallelize TestContextCancelStmtQuery

* parallelize TestContextCancelBegin

* parallelize TestContextBeginIsolationLevel

* parallelize TestContextBeginReadOnly

* parallelize TestValuerWithValueReceiverGivenNilValue

* parallelize TestRawBytesAreNotModified

* parallelize TestFoundRows

* parallelize TestRowsClose

* parallelize TestCloseStmtBeforeRows

* parallelize TestStmtMultiRows

* Revert "parallelize TestRawBytesAreNotModified"

This reverts commit 91622f05d.

* parallelize TestStaleConnectionChecks

* parallelize TestFailingCharset

* parallelize TestColumnsWithAlias

* parallelize TestRawBytesResultExceedsBuffer

* parallelize TestUnixSocketAuthFail

* parallelize TestSkipResults

* Add parallel flag to go test command

* Revert "parallelize TestUnixSocketAuthFail"

This reverts commit b3df7bd13.

Run Details

3087 of 3758 relevant lines covered (82.14%)

1304577.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.44

/connection.go

// Go MySQL Driver - A MySQL-Driver for Go's database/sql package
//
// Copyright 2012 The Go-MySQL-Driver Authors. All rights reserved.
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
// You can obtain one at http://mozilla.org/MPL/2.0/.

package mysql

import (
        "context"
        "database/sql"
        "database/sql/driver"
        "encoding/json"
        "io"
        "net"
        "strconv"
        "strings"
        "time"
)

type mysqlConn struct {
        buf              buffer
        netConn          net.Conn
        rawConn          net.Conn    // underlying connection when netConn is TLS connection.
        result           mysqlResult // managed by clearResult() and handleOkPacket().
        cfg              *Config
        connector        *connector
        maxAllowedPacket int
        maxWriteSize     int
        writeTimeout     time.Duration
        flags            clientFlag
        status           statusFlag
        sequence         uint8
        parseTime        bool

        // for context support (Go 1.8+)
        watching bool
        watcher  chan<- context.Context
        closech  chan struct{}
        finished chan<- struct{}
        canceled atomicError // set non-nil if conn is canceled
        closed   atomicBool  // set when conn is closed, before closech is closed
}

// Handles parameters set in DSN after the connection is established
func (mc *mysqlConn) handleParams() (err error) {
        var cmdSet strings.Builder

        for param, val := range mc.cfg.Params {
                switch param {
                // Charset: character_set_connection, character_set_client, character_set_results
                case "charset":
                        charsets := strings.Split(val, ",")
                        for _, cs := range charsets {
                                // ignore errors here - a charset may not exist
                                if mc.cfg.Collation != "" {
                                        err = mc.exec("SET NAMES " + cs + " COLLATE " + mc.cfg.Collation)
                                } else {
                                        err = mc.exec("SET NAMES " + cs)
                                }
                                if err == nil {
                                        break
                                }
                        }
                        if err != nil {
                                return
                        }

                // Other system vars accumulated in a single SET command
                default:
                        if cmdSet.Len() == 0 {
                                // Heuristic: 29 chars for each other key=value to reduce reallocations
                                cmdSet.Grow(4 + len(param) + 3 + len(val) + 30*(len(mc.cfg.Params)-1))
                                cmdSet.WriteString("SET ")
                        } else {
                                cmdSet.WriteString(", ")
                        }
                        cmdSet.WriteString(param)
                        cmdSet.WriteString(" = ")
                        cmdSet.WriteString(val)
                }
        }

        if cmdSet.Len() > 0 {
                err = mc.exec(cmdSet.String())
                if err != nil {
                        return
                }
        }

        return
}

func (mc *mysqlConn) markBadConn(err error) error {
        if mc == nil {
                return err
        }
        if err != errBadConnNoWrite {
                return err
        }
        return driver.ErrBadConn
}

func (mc *mysqlConn) Begin() (driver.Tx, error) {
        return mc.begin(false)
}

func (mc *mysqlConn) begin(readOnly bool) (driver.Tx, error) {
        if mc.closed.Load() {
                mc.cfg.Logger.Print(ErrInvalidConn)
                return nil, driver.ErrBadConn
        }
        var q string
        if readOnly {
                q = "START TRANSACTION READ ONLY"
        } else {
                q = "START TRANSACTION"
        }
        err := mc.exec(q)
        if err == nil {
                return &mysqlTx{mc}, err
        }
        return nil, mc.markBadConn(err)
}

func (mc *mysqlConn) Close() (err error) {
        // Makes Close idempotent
        if !mc.closed.Load() {
                err = mc.writeCommandPacket(comQuit)
        }

        mc.cleanup()

        return
}

// Closes the network connection and unsets internal variables. Do not call this
// function after successfully authentication, call Close instead. This function
// is called before auth or on auth failure because MySQL will have already
// closed the network connection.
func (mc *mysqlConn) cleanup() {
        if mc.closed.Swap(true) {
                return
        }

        // Makes cleanup idempotent
        close(mc.closech)
        if mc.netConn == nil {
                return
        }
        if err := mc.netConn.Close(); err != nil {
                mc.cfg.Logger.Print(err)
        }
        mc.clearResult()
}

func (mc *mysqlConn) error() error {
        if mc.closed.Load() {
                if err := mc.canceled.Value(); err != nil {
                        return err
                }
                return ErrInvalidConn
        }
        return nil
}

func (mc *mysqlConn) Prepare(query string) (driver.Stmt, error) {
        if mc.closed.Load() {
                mc.cfg.Logger.Print(ErrInvalidConn)
                return nil, driver.ErrBadConn
        }
        // Send command
        err := mc.writeCommandPacketStr(comStmtPrepare, query)
        if err != nil {
                // STMT_PREPARE is safe to retry.  So we can return ErrBadConn here.
                mc.cfg.Logger.Print(err)
                return nil, driver.ErrBadConn
        }

        stmt := &mysqlStmt{
                mc: mc,
        }

        // Read Result
        columnCount, err := stmt.readPrepareResultPacket()
        if err == nil {
                if stmt.paramCount > 0 {
                        if err = mc.readUntilEOF(); err != nil {
                                return nil, err
                        }
                }

                if columnCount > 0 {
                        err = mc.readUntilEOF()
                }
        }

        return stmt, err
}

func (mc *mysqlConn) interpolateParams(query string, args []driver.Value) (string, error) {
        // Number of ? should be same to len(args)
        if strings.Count(query, "?") != len(args) {
                return "", driver.ErrSkip
        }

        buf, err := mc.buf.takeCompleteBuffer()
        if err != nil {
                // can not take the buffer. Something must be wrong with the connection
                mc.cfg.Logger.Print(err)
                return "", ErrInvalidConn
        }
        buf = buf[:0]
        argPos := 0

        for i := 0; i < len(query); i++ {
                q := strings.IndexByte(query[i:], '?')
                if q == -1 {
                        buf = append(buf, query[i:]...)
                        break
                }
                buf = append(buf, query[i:i+q]...)
                i += q

                arg := args[argPos]
                argPos++

                if arg == nil {
                        buf = append(buf, "NULL"...)
                        continue
                }

                switch v := arg.(type) {
                case int64:
                        buf = strconv.AppendInt(buf, v, 10)
                case uint64:
                        // Handle uint64 explicitly because our custom ConvertValue emits unsigned values
                        buf = strconv.AppendUint(buf, v, 10)
                case float64:
                        buf = strconv.AppendFloat(buf, v, 'g', -1, 64)
                case bool:
                        if v {
                                buf = append(buf, '1')
                        } else {
                                buf = append(buf, '0')
                        }
                case time.Time:
                        if v.IsZero() {
                                buf = append(buf, "'0000-00-00'"...)
                        } else {
                                buf = append(buf, '\'')
                                buf, err = appendDateTime(buf, v.In(mc.cfg.Loc))
                                if err != nil {
                                        return "", err
                                }
                                buf = append(buf, '\'')
                        }
                case json.RawMessage:
                        buf = append(buf, '\'')
                        if mc.status&statusNoBackslashEscapes == 0 {
                                buf = escapeBytesBackslash(buf, v)
                        } else {
                                buf = escapeBytesQuotes(buf, v)
                        }
                        buf = append(buf, '\'')
                case []byte:
                        if v == nil {
                                buf = append(buf, "NULL"...)
                        } else {
                                buf = append(buf, "_binary'"...)
                                if mc.status&statusNoBackslashEscapes == 0 {
                                        buf = escapeBytesBackslash(buf, v)
                                } else {
                                        buf = escapeBytesQuotes(buf, v)
                                }
                                buf = append(buf, '\'')
                        }
                case string:
                        buf = append(buf, '\'')
                        if mc.status&statusNoBackslashEscapes == 0 {
                                buf = escapeStringBackslash(buf, v)
                        } else {
                                buf = escapeStringQuotes(buf, v)
                        }
                        buf = append(buf, '\'')
                default:
                        return "", driver.ErrSkip
                }

                if len(buf)+4 > mc.maxAllowedPacket {
                        return "", driver.ErrSkip
                }
        }
        if argPos != len(args) {
                return "", driver.ErrSkip
        }
        return string(buf), nil
}

func (mc *mysqlConn) Exec(query string, args []driver.Value) (driver.Result, error) {
        if mc.closed.Load() {
                mc.cfg.Logger.Print(ErrInvalidConn)
                return nil, driver.ErrBadConn
        }
        if len(args) != 0 {
                if !mc.cfg.InterpolateParams {
                        return nil, driver.ErrSkip
                }
                // try to interpolate the parameters to save extra roundtrips for preparing and closing a statement
                prepared, err := mc.interpolateParams(query, args)
                if err != nil {
                        return nil, err
                }
                query = prepared
        }

        err := mc.exec(query)
        if err == nil {
                copied := mc.result
                return &copied, err
        }
        return nil, mc.markBadConn(err)
}

// Internal function to execute commands
func (mc *mysqlConn) exec(query string) error {
        handleOk := mc.clearResult()
        // Send command
        if err := mc.writeCommandPacketStr(comQuery, query); err != nil {
                return mc.markBadConn(err)
        }

        // Read Result
        resLen, err := handleOk.readResultSetHeaderPacket()
        if err != nil {
                return err
        }

        if resLen > 0 {
                // columns
                if err := mc.readUntilEOF(); err != nil {
                        return err
                }

                // rows
                if err := mc.readUntilEOF(); err != nil {
                        return err
                }
        }

        return handleOk.discardResults()
}

func (mc *mysqlConn) Query(query string, args []driver.Value) (driver.Rows, error) {
        return mc.query(query, args)
}

func (mc *mysqlConn) query(query string, args []driver.Value) (*textRows, error) {
        handleOk := mc.clearResult()

        if mc.closed.Load() {
                mc.cfg.Logger.Print(ErrInvalidConn)
                return nil, driver.ErrBadConn
        }
        if len(args) != 0 {
                if !mc.cfg.InterpolateParams {
                        return nil, driver.ErrSkip
                }
                // try client-side prepare to reduce roundtrip
                prepared, err := mc.interpolateParams(query, args)
                if err != nil {
                        return nil, err
                }
                query = prepared
        }
        // Send command
        err := mc.writeCommandPacketStr(comQuery, query)
        if err == nil {
                // Read Result
                var resLen int
                resLen, err = handleOk.readResultSetHeaderPacket()
                if err == nil {
                        rows := new(textRows)
                        rows.mc = mc

                        if resLen == 0 {
                                rows.rs.done = true

                                switch err := rows.NextResultSet(); err {
                                case nil, io.EOF:
                                        return rows, nil
                                default:
                                        return nil, err
                                }
                        }

                        // Columns
                        rows.rs.columns, err = mc.readColumns(resLen)
                        return rows, err
                }
        }
        return nil, mc.markBadConn(err)
}

// Gets the value of the given MySQL System Variable
// The returned byte slice is only valid until the next read
func (mc *mysqlConn) getSystemVar(name string) ([]byte, error) {
        // Send command
        handleOk := mc.clearResult()
        if err := mc.writeCommandPacketStr(comQuery, "SELECT @@"+name); err != nil {
                return nil, err
        }

        // Read Result
        resLen, err := handleOk.readResultSetHeaderPacket()
        if err == nil {
                rows := new(textRows)
                rows.mc = mc
                rows.rs.columns = []mysqlField{{fieldType: fieldTypeVarChar}}

                if resLen > 0 {
                        // Columns
                        if err := mc.readUntilEOF(); err != nil {
                                return nil, err
                        }
                }

                dest := make([]driver.Value, resLen)
                if err = rows.readRow(dest); err == nil {
                        return dest[0].([]byte), mc.readUntilEOF()
                }
        }
        return nil, err
}

// finish is called when the query has canceled.
func (mc *mysqlConn) cancel(err error) {
        mc.canceled.Set(err)
        mc.cleanup()
}

// finish is called when the query has succeeded.
func (mc *mysqlConn) finish() {
        if !mc.watching || mc.finished == nil {
                return
        }
        select {
        case mc.finished <- struct{}{}:
                mc.watching = false
        case <-mc.closech:
        }
}

// Ping implements driver.Pinger interface
func (mc *mysqlConn) Ping(ctx context.Context) (err error) {
        if mc.closed.Load() {
                mc.cfg.Logger.Print(ErrInvalidConn)
                return driver.ErrBadConn
        }

        if err = mc.watchCancel(ctx); err != nil {
                return
        }
        defer mc.finish()

        handleOk := mc.clearResult()
        if err = mc.writeCommandPacket(comPing); err != nil {
                return mc.markBadConn(err)
        }

        return handleOk.readResultOK()
}

// BeginTx implements driver.ConnBeginTx interface
func (mc *mysqlConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
        if mc.closed.Load() {
                return nil, driver.ErrBadConn
        }

        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }
        defer mc.finish()

        if sql.IsolationLevel(opts.Isolation) != sql.LevelDefault {
                level, err := mapIsolationLevel(opts.Isolation)
                if err != nil {
                        return nil, err
                }
                err = mc.exec("SET TRANSACTION ISOLATION LEVEL " + level)
                if err != nil {
                        return nil, err
                }
        }

        return mc.begin(opts.ReadOnly)
}

func (mc *mysqlConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }

        rows, err := mc.query(query, dargs)
        if err != nil {
                mc.finish()
                return nil, err
        }
        rows.finish = mc.finish
        return rows, err
}

func (mc *mysqlConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }
        defer mc.finish()

        return mc.Exec(query, dargs)
}

func (mc *mysqlConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }

        stmt, err := mc.Prepare(query)
        mc.finish()
        if err != nil {
                return nil, err
        }

        select {
        default:
        case <-ctx.Done():
                stmt.Close()
                return nil, ctx.Err()
        }
        return stmt, nil
}

func (stmt *mysqlStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := stmt.mc.watchCancel(ctx); err != nil {
                return nil, err
        }

        rows, err := stmt.query(dargs)
        if err != nil {
                stmt.mc.finish()
                return nil, err
        }
        rows.finish = stmt.mc.finish
        return rows, err
}

func (stmt *mysqlStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := stmt.mc.watchCancel(ctx); err != nil {
                return nil, err
        }
        defer stmt.mc.finish()

        return stmt.Exec(dargs)
}

func (mc *mysqlConn) watchCancel(ctx context.Context) error {
        if mc.watching {
                // Reach here if canceled,
                // so the connection is already invalid
                mc.cleanup()
                return nil
        }
        // When ctx is already cancelled, don't watch it.
        if err := ctx.Err(); err != nil {
                return err
        }
        // When ctx is not cancellable, don't watch it.
        if ctx.Done() == nil {
                return nil
        }
        // When watcher is not alive, can't watch it.
        if mc.watcher == nil {
                return nil
        }

        mc.watching = true
        mc.watcher <- ctx
        return nil
}

func (mc *mysqlConn) startWatcher() {
        watcher := make(chan context.Context, 1)
        mc.watcher = watcher
        finished := make(chan struct{})
        mc.finished = finished
        go func() {
                for {
                        var ctx context.Context
                        select {
                        case ctx = <-watcher:
                        case <-mc.closech:
                                return
                        }

                        select {
                        case <-ctx.Done():
                                mc.cancel(ctx.Err())
                        case <-finished:
                        case <-mc.closech:
                                return
                        }
                }
        }()
}

func (mc *mysqlConn) CheckNamedValue(nv *driver.NamedValue) (err error) {
        nv.Value, err = converter{}.ConvertValue(nv.Value)
        return
}

// ResetSession implements driver.SessionResetter.
// (From Go 1.10)
func (mc *mysqlConn) ResetSession(ctx context.Context) error {
        if mc.closed.Load() {
                return driver.ErrBadConn
        }

        // Perform a stale connection check. We only perform this check for
        // the first query on a connection that has been checked out of the
        // connection pool: a fresh connection from the pool is more likely
        // to be stale, and it has not performed any previous writes that
        // could cause data corruption, so it's safe to return ErrBadConn
        // if the check fails.
        if mc.cfg.CheckConnLiveness {
                conn := mc.netConn
                if mc.rawConn != nil {
                        conn = mc.rawConn
                }
                var err error
                if mc.cfg.ReadTimeout != 0 {
                        err = conn.SetReadDeadline(time.Now().Add(mc.cfg.ReadTimeout))
                }
                if err == nil {
                        err = connCheck(conn)
                }
                if err != nil {
                        mc.cfg.Logger.Print("closing bad idle connection: ", err)
                        return driver.ErrBadConn
                }
        }

        return nil
}

// IsValid implements driver.Validator interface
// (From Go 1.15)
func (mc *mysqlConn) IsValid() bool {
        return !mc.closed.Load()
}

1	// Go MySQL Driver - A MySQL-Driver for Go's database/sql package
2	//
3	// Copyright 2012 The Go-MySQL-Driver Authors. All rights reserved.
4	//
5	// This Source Code Form is subject to the terms of the Mozilla Public
6	// License, v. 2.0. If a copy of the MPL was not distributed with this file,
7	// You can obtain one at http://mozilla.org/MPL/2.0/.
8
9	package mysql
10
11	import (
12	"context"
13	"database/sql"
14	"database/sql/driver"
15	"encoding/json"
16	"io"
17	"net"
18	"strconv"
19	"strings"
20	"time"
21	)
22
23	type mysqlConn struct {
24	buf buffer
25	netConn net.Conn
26	rawConn net.Conn // underlying connection when netConn is TLS connection.
27	result mysqlResult // managed by clearResult() and handleOkPacket().
28	cfg *Config
29	connector *connector
30	maxAllowedPacket int
31	maxWriteSize int
32	writeTimeout time.Duration
33	flags clientFlag
34	status statusFlag
35	sequence uint8
36	parseTime bool
37
38	// for context support (Go 1.8+)
39	watching bool
40	watcher chan<- context.Context
41	closech chan struct{}
42	finished chan<- struct{}
43	canceled atomicError // set non-nil if conn is canceled
44	closed atomicBool // set when conn is closed, before closech is closed
45	}
46
47	// Handles parameters set in DSN after the connection is established
48	func (mc *mysqlConn) handleParams() (err error) {	8,070✔
49	var cmdSet strings.Builder	8,070✔
50		8,070✔
51	for param, val := range mc.cfg.Params {	8,610✔
52	switch param {	540✔
53	// Charset: character_set_connection, character_set_client, character_set_results
54	case "charset":	300✔
55	charsets := strings.Split(val, ",")	300✔
56	for _, cs := range charsets {	660✔
57	// ignore errors here - a charset may not exist	360✔
58	if mc.cfg.Collation != "" {	360✔
59	err = mc.exec("SET NAMES " + cs + " COLLATE " + mc.cfg.Collation)	×
60	} else {	360✔
61	err = mc.exec("SET NAMES " + cs)	360✔
62	}	360✔
63	if err == nil {	600✔
64	break	240✔
65	}
66	}
67	if err != nil {	360✔
68	return	60✔
69	}	60✔
70
71	// Other system vars accumulated in a single SET command
72	default:	240✔
73	if cmdSet.Len() == 0 {	480✔
74	// Heuristic: 29 chars for each other key=value to reduce reallocations	240✔
75	cmdSet.Grow(4 + len(param) + 3 + len(val) + 30*(len(mc.cfg.Params)-1))	240✔
76	cmdSet.WriteString("SET ")	240✔
77	} else {	240✔
78	cmdSet.WriteString(", ")	×
79	}	×
80	cmdSet.WriteString(param)	240✔
81	cmdSet.WriteString(" = ")	240✔
82	cmdSet.WriteString(val)	240✔
83	}
84	}
85
86	if cmdSet.Len() > 0 {	8,250✔
87	err = mc.exec(cmdSet.String())	240✔
88	if err != nil {	240✔
89	return	×
90	}	×
91	}
92
93	return	8,010✔
94	}
95
96	func (mc *mysqlConn) markBadConn(err error) error {	438✔
97	if mc == nil {	438✔
98	return err	×
99	}	×
100	if err != errBadConnNoWrite {	846✔
101	return err	408✔
102	}	408✔
103	return driver.ErrBadConn	30✔
104	}
105
106	func (mc *mysqlConn) Begin() (driver.Tx, error) {	×
107	return mc.begin(false)	×
108	}	×
109
110	func (mc *mysqlConn) begin(readOnly bool) (driver.Tx, error) {	3,112✔
111	if mc.closed.Load() {	3,112✔
112	mc.cfg.Logger.Print(ErrInvalidConn)	×
113	return nil, driver.ErrBadConn	×
114	}	×
115	var q string	3,112✔
116	if readOnly {	3,142✔
117	q = "START TRANSACTION READ ONLY"	30✔
118	} else {	3,112✔
119	q = "START TRANSACTION"	3,082✔
120	}	3,082✔
121	err := mc.exec(q)	3,112✔
122	if err == nil {	6,224✔
123	return &mysqlTx{mc}, err	3,112✔
124	}	3,112✔
125	return nil, mc.markBadConn(err)	×
126	}
127
128	func (mc *mysqlConn) Close() (err error) {	8,130✔
129	// Makes Close idempotent	8,130✔
130	if !mc.closed.Load() {	14,778✔
131	err = mc.writeCommandPacket(comQuit)	6,648✔
132	}	6,648✔
133
134	mc.cleanup()	8,130✔
135		8,130✔
136	return	8,130✔
137	}
138
139	// Closes the network connection and unsets internal variables. Do not call this
140	// function after successfully authentication, call Close instead. This function
141	// is called before auth or on auth failure because MySQL will have already
142	// closed the network connection.
143	func (mc *mysqlConn) cleanup() {	9,725✔
144	if mc.closed.Swap(true) {	11,237✔
145	return	1,512✔
146	}	1,512✔
147
148	// Makes cleanup idempotent
149	close(mc.closech)	8,213✔
150	if mc.netConn == nil {	8,273✔
151	return	60✔
152	}	60✔
153	if err := mc.netConn.Close(); err != nil {	8,153✔
154	mc.cfg.Logger.Print(err)	×
155	}	×
156	mc.clearResult()	8,153✔
157	}
158
159	func (mc *mysqlConn) error() error {	17,244✔
160	if mc.closed.Load() {	18,474✔
161	if err := mc.canceled.Value(); err != nil {	2,460✔
162	return err	1,230✔
163	}	1,230✔
164	return ErrInvalidConn	×
165	}
166	return nil	16,014✔
167	}
168
169	func (mc *mysqlConn) Prepare(query string) (driver.Stmt, error) {	2,760✔
170	if mc.closed.Load() {	2,760✔
171	mc.cfg.Logger.Print(ErrInvalidConn)	×
172	return nil, driver.ErrBadConn	×
173	}	×
174	// Send command
175	err := mc.writeCommandPacketStr(comStmtPrepare, query)	2,760✔
176	if err != nil {	2,760✔
177	// STMT_PREPARE is safe to retry. So we can return ErrBadConn here.	×
178	mc.cfg.Logger.Print(err)	×
179	return nil, driver.ErrBadConn	×
180	}	×
181
182	stmt := &mysqlStmt{	2,760✔
183	mc: mc,	2,760✔
184	}	2,760✔
185		2,760✔
186	// Read Result	2,760✔
187	columnCount, err := stmt.readPrepareResultPacket()	2,760✔
188	if err == nil {	5,520✔
189	if stmt.paramCount > 0 {	5,220✔
190	if err = mc.readUntilEOF(); err != nil {	2,460✔
191	return nil, err	×
192	}	×
193	}
194
195	if columnCount > 0 {	4,080✔
196	err = mc.readUntilEOF()	1,320✔
197	}	1,320✔
198	}
199
200	return stmt, err	2,760✔
201	}
202
203	func (mc *mysqlConn) interpolateParams(query string, args []driver.Value) (string, error) {	1,410✔
204	// Number of ? should be same to len(args)	1,410✔
205	if strings.Count(query, "?") != len(args) {	1,470✔
206	return "", driver.ErrSkip	60✔
207	}	60✔
208
209	buf, err := mc.buf.takeCompleteBuffer()	1,350✔
210	if err != nil {	1,350✔
211	// can not take the buffer. Something must be wrong with the connection	×
212	mc.cfg.Logger.Print(err)	×
213	return "", ErrInvalidConn	×
214	}	×
215	buf = buf[:0]	1,350✔
216	argPos := 0	1,350✔
217		1,350✔
218	for i := 0; i < len(query); i++ {	3,930✔
219	q := strings.IndexByte(query[i:], '?')	2,580✔
220	if q == -1 {	3,660✔
221	buf = append(buf, query[i:]...)	1,080✔
222	break	1,080✔
223	}
224	buf = append(buf, query[i:i+q]...)	1,500✔
225	i += q	1,500✔
226		1,500✔
227	arg := args[argPos]	1,500✔
228	argPos++	1,500✔
229		1,500✔
230	if arg == nil {	1,500✔
231	buf = append(buf, "NULL"...)	×
232	continue	×
233	}
234
235	switch v := arg.(type) {	1,500✔
236	case int64:	300✔
237	buf = strconv.AppendInt(buf, v, 10)	300✔
238	case uint64:	30✔
239	// Handle uint64 explicitly because our custom ConvertValue emits unsigned values	30✔
240	buf = strconv.AppendUint(buf, v, 10)	30✔
241	case float64:	×
242	buf = strconv.AppendFloat(buf, v, 'g', -1, 64)	×
243	case bool:	×
244	if v {	×
245	buf = append(buf, '1')	×
246	} else {	×
247	buf = append(buf, '0')	×
248	}	×
249	case time.Time:	360✔
250	if v.IsZero() {	420✔
251	buf = append(buf, "'0000-00-00'"...)	60✔
252	} else {	360✔
253	buf = append(buf, '\'')	300✔
254	buf, err = appendDateTime(buf, v.In(mc.cfg.Loc))	300✔
255	if err != nil {	300✔
256	return "", err	×
257	}	×
258	buf = append(buf, '\'')	300✔
259	}
260	case json.RawMessage:	30✔
261	buf = append(buf, '\'')	30✔
262	if mc.status&statusNoBackslashEscapes == 0 {	60✔
263	buf = escapeBytesBackslash(buf, v)	30✔
264	} else {	30✔
265	buf = escapeBytesQuotes(buf, v)	×
266	}	×
267	buf = append(buf, '\'')	30✔
268	case []byte:	×
269	if v == nil {	×
270	buf = append(buf, "NULL"...)	×
271	} else {	×
272	buf = append(buf, "_binary'"...)	×
273	if mc.status&statusNoBackslashEscapes == 0 {	×
274	buf = escapeBytesBackslash(buf, v)	×
275	} else {	×
276	buf = escapeBytesQuotes(buf, v)	×
277	}	×
278	buf = append(buf, '\'')	×
279	}
280	case string:	780✔
281	buf = append(buf, '\'')	780✔
282	if mc.status&statusNoBackslashEscapes == 0 {	1,470✔
283	buf = escapeStringBackslash(buf, v)	690✔
284	} else {	780✔
285	buf = escapeStringQuotes(buf, v)	90✔
286	}	90✔
287	buf = append(buf, '\'')	780✔
288	default:	×
289	return "", driver.ErrSkip	×
290	}
291
292	if len(buf)+4 > mc.maxAllowedPacket {	1,530✔
293	return "", driver.ErrSkip	30✔
294	}	30✔
295	}
296	if argPos != len(args) {	1,320✔
297	return "", driver.ErrSkip	×
298	}	×
299	return string(buf), nil	1,320✔
300	}
301
302	func (mc *mysqlConn) Exec(query string, args []driver.Value) (driver.Result, error) {	13,098✔
303	if mc.closed.Load() {	13,098✔
304	mc.cfg.Logger.Print(ErrInvalidConn)	×
305	return nil, driver.ErrBadConn	×
306	}	×
307	if len(args) != 0 {	14,778✔
308	if !mc.cfg.InterpolateParams {	2,910✔
309	return nil, driver.ErrSkip	1,230✔
310	}	1,230✔
311	// try to interpolate the parameters to save extra roundtrips for preparing and closing a statement
312	prepared, err := mc.interpolateParams(query, args)	450✔
313	if err != nil {	480✔
314	return nil, err	30✔
315	}	30✔
316	query = prepared	420✔
317	}
318
319	err := mc.exec(query)	11,838✔
320	if err == nil {	23,364✔
321	copied := mc.result	11,526✔
322	return &copied, err	11,526✔
323	}	11,526✔
324	return nil, mc.markBadConn(err)	312✔
325	}
326
327	// Internal function to execute commands
328	func (mc *mysqlConn) exec(query string) error {	18,710✔
329	handleOk := mc.clearResult()	18,710✔
330	// Send command	18,710✔
331	if err := mc.writeCommandPacketStr(comQuery, query); err != nil {	18,710✔
332	return mc.markBadConn(err)	×
333	}	×
334
335	// Read Result
336	resLen, err := handleOk.readResultSetHeaderPacket()	18,710✔
337	if err != nil {	19,142✔
338	return err	432✔
339	}	432✔
340
341	if resLen > 0 {	18,278✔
342	// columns	×
343	if err := mc.readUntilEOF(); err != nil {	×
344	return err	×
345	}	×
346
347	// rows
348	if err := mc.readUntilEOF(); err != nil {	×
349	return err	×
350	}	×
351	}
352
353	return handleOk.discardResults()	18,278✔
354	}
355
356	func (mc *mysqlConn) Query(query string, args []driver.Value) (driver.Rows, error) {	120✔
357	return mc.query(query, args)	120✔
358	}	120✔
359
360	func (mc mysqlConn) query(query string, args []driver.Value) (textRows, error) {	7,380✔
361	handleOk := mc.clearResult()	7,380✔
362		7,380✔
363	if mc.closed.Load() {	7,380✔
364	mc.cfg.Logger.Print(ErrInvalidConn)	×
365	return nil, driver.ErrBadConn	×
366	}	×
367	if len(args) != 0 {	9,300✔
368	if !mc.cfg.InterpolateParams {	3,030✔
369	return nil, driver.ErrSkip	1,110✔
370	}	1,110✔
371	// try client-side prepare to reduce roundtrip
372	prepared, err := mc.interpolateParams(query, args)	810✔
373	if err != nil {	810✔
374	return nil, err	×
375	}	×
376	query = prepared	810✔
377	}
378	// Send command
379	err := mc.writeCommandPacketStr(comQuery, query)	6,270✔
380	if err == nil {	12,540✔
381	// Read Result	6,270✔
382	var resLen int	6,270✔
383	resLen, err = handleOk.readResultSetHeaderPacket()	6,270✔
384	if err == nil {	12,474✔
385	rows := new(textRows)	6,204✔
386	rows.mc = mc	6,204✔
387		6,204✔
388	if resLen == 0 {	6,354✔
389	rows.rs.done = true	150✔
390		150✔
391	switch err := rows.NextResultSet(); err {	150✔
392	case nil, io.EOF:	150✔
393	return rows, nil	150✔
394	default:	×
395	return nil, err	×
396	}
397	}
398
399	// Columns
400	rows.rs.columns, err = mc.readColumns(resLen)	6,054✔
401	return rows, err	6,054✔
402	}
403	}
404	return nil, mc.markBadConn(err)	66✔
405	}
406
407	// Gets the value of the given MySQL System Variable
408	// The returned byte slice is only valid until the next read
409	func (mc *mysqlConn) getSystemVar(name string) ([]byte, error) {	60✔
410	// Send command	60✔
411	handleOk := mc.clearResult()	60✔
412	if err := mc.writeCommandPacketStr(comQuery, "SELECT @@"+name); err != nil {	60✔
413	return nil, err	×
414	}	×
415
416	// Read Result
417	resLen, err := handleOk.readResultSetHeaderPacket()	60✔
418	if err == nil {	120✔
419	rows := new(textRows)	60✔
420	rows.mc = mc	60✔
421	rows.rs.columns = []mysqlField{{fieldType: fieldTypeVarChar}}	60✔
422		60✔
423	if resLen > 0 {	120✔
424	// Columns	60✔
425	if err := mc.readUntilEOF(); err != nil {	60✔
426	return nil, err	×
427	}	×
428	}
429
430	dest := make([]driver.Value, resLen)	60✔
431	if err = rows.readRow(dest); err == nil {	120✔
432	return dest[0].([]byte), mc.readUntilEOF()	60✔
433	}	60✔
434	}
435	return nil, err	×
436	}
437
438	// finish is called when the query has canceled.
439	func (mc *mysqlConn) cancel(err error) {	1,392✔
440	mc.canceled.Set(err)	1,392✔
441	mc.cleanup()	1,392✔
442	}	1,392✔
443
444	// finish is called when the query has succeeded.
445	func (mc *mysqlConn) finish() {	37,884✔
446	if !mc.watching \|\| mc.finished == nil {	72,894✔
447	return	35,010✔
448	}	35,010✔
449	select {	2,874✔
450	case mc.finished <- struct{}{}:	1,482✔
451	mc.watching = false	1,482✔
452	case <-mc.closech:	1,392✔
453	}
454	}
455
456	// Ping implements driver.Pinger interface
457	func (mc *mysqlConn) Ping(ctx context.Context) (err error) {	561✔
458	if mc.closed.Load() {	561✔
459	mc.cfg.Logger.Print(ErrInvalidConn)	×
460	return driver.ErrBadConn	×
461	}	×
462
463	if err = mc.watchCancel(ctx); err != nil {	651✔
464	return	90✔
465	}	90✔
466	defer mc.finish()	471✔
467		471✔
468	handleOk := mc.clearResult()	471✔
469	if err = mc.writeCommandPacket(comPing); err != nil {	531✔
470	return mc.markBadConn(err)	60✔
471	}	60✔
472
473	return handleOk.readResultOK()	411✔
474	}
475
476	// BeginTx implements driver.ConnBeginTx interface
477	func (mc *mysqlConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {	3,124✔
478	if mc.closed.Load() {	3,136✔
479	return nil, driver.ErrBadConn	12✔
480	}	12✔
481
482	if err := mc.watchCancel(ctx); err != nil {	3,112✔
483	return nil, err	×
484	}	×
485	defer mc.finish()	3,112✔
486		3,112✔
487	if sql.IsolationLevel(opts.Isolation) != sql.LevelDefault {	3,172✔
488	level, err := mapIsolationLevel(opts.Isolation)	60✔
489	if err != nil {	60✔
490	return nil, err	×
491	}	×
492	err = mc.exec("SET TRANSACTION ISOLATION LEVEL " + level)	60✔
493	if err != nil {	60✔
494	return nil, err	×
495	}	×
496	}
497
498	return mc.begin(opts.ReadOnly)	3,112✔
499	}
500
501	func (mc *mysqlConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {	7,260✔
502	dargs, err := namedValueToValue(args)	7,260✔
503	if err != nil {	7,260✔
504	return nil, err	×
505	}	×
506
507	if err := mc.watchCancel(ctx); err != nil {	7,260✔
508	return nil, err	×
509	}	×
510
511	rows, err := mc.query(query, dargs)	7,260✔
512	if err != nil {	8,436✔
513	mc.finish()	1,176✔
514	return nil, err	1,176✔
515	}	1,176✔
516	rows.finish = mc.finish	6,084✔
517	return rows, err	6,084✔
518	}
519
520	func (mc *mysqlConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {	13,068✔
521	dargs, err := namedValueToValue(args)	13,068✔
522	if err != nil {	13,068✔
523	return nil, err	×
524	}	×
525
526	if err := mc.watchCancel(ctx); err != nil {	13,068✔
527	return nil, err	×
528	}	×
529	defer mc.finish()	13,068✔
530		13,068✔
531	return mc.Exec(query, dargs)	13,068✔
532	}
533
534	func (mc *mysqlConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {	2,730✔
535	if err := mc.watchCancel(ctx); err != nil {	2,730✔
536	return nil, err	×
537	}	×
538
539	stmt, err := mc.Prepare(query)	2,730✔
540	mc.finish()	2,730✔
541	if err != nil {	2,730✔
542	return nil, err	×
543	}	×
544
545	select {	2,730✔
546	default:	2,730✔
547	case <-ctx.Done():	×
548	stmt.Close()	×
549	return nil, ctx.Err()	×
550	}
551	return stmt, nil	2,730✔
552	}
553
554	func (stmt *mysqlStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {	1,800✔
555	dargs, err := namedValueToValue(args)	1,800✔
556	if err != nil {	1,800✔
557	return nil, err	×
558	}	×
559
560	if err := stmt.mc.watchCancel(ctx); err != nil {	1,800✔
561	return nil, err	×
562	}	×
563
564	rows, err := stmt.query(dargs)	1,800✔
565	if err != nil {	1,830✔
566	stmt.mc.finish()	30✔
567	return nil, err	30✔
568	}	30✔
569	rows.finish = stmt.mc.finish	1,770✔
570	return rows, err	1,770✔
571	}
572
573	func (stmt *mysqlStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {	1,290✔
574	dargs, err := namedValueToValue(args)	1,290✔
575	if err != nil {	1,290✔
576	return nil, err	×
577	}	×
578
579	if err := stmt.mc.watchCancel(ctx); err != nil {	1,290✔
580	return nil, err	×
581	}	×
582	defer stmt.mc.finish()	1,290✔
583		1,290✔
584	return stmt.Exec(dargs)	1,290✔
585	}
586
587	func (mc *mysqlConn) watchCancel(ctx context.Context) error {	38,034✔
588	if mc.watching {	38,034✔
589	// Reach here if canceled,	×
590	// so the connection is already invalid	×
591	mc.cleanup()	×
592	return nil	×
593	}	×
594	// When ctx is already cancelled, don't watch it.
595	if err := ctx.Err(); err != nil {	38,154✔
596	return err	120✔
597	}	120✔
598	// When ctx is not cancellable, don't watch it.
599	if ctx.Done() == nil {	72,954✔
600	return nil	35,040✔
601	}	35,040✔
602	// When watcher is not alive, can't watch it.
603	if mc.watcher == nil {	2,874✔
604	return nil	×
605	}	×
606
607	mc.watching = true	2,874✔
608	mc.watcher <- ctx	2,874✔
609	return nil	2,874✔
610	}
611
612	func (mc *mysqlConn) startWatcher() {	8,243✔
613	watcher := make(chan context.Context, 1)	8,243✔
614	mc.watcher = watcher	8,243✔
615	finished := make(chan struct{})	8,243✔
616	mc.finished = finished	8,243✔
617	go func() {	16,486✔
618	for {	19,360✔
619	var ctx context.Context	11,117✔
620	select {	11,117✔
621	case ctx = <-watcher:	2,874✔
622	case <-mc.closech:	8,093✔
623	return	8,093✔
624	}
625
626	select {	2,874✔
627	case <-ctx.Done():	1,392✔
628	mc.cancel(ctx.Err())	1,392✔
629	case <-finished:	1,482✔
630	case <-mc.closech:	×
631	return	×
632	}
633	}
634	}()
635	}
636
637	func (mc mysqlConn) CheckNamedValue(nv driver.NamedValue) (err error) {	4,110✔
638	nv.Value, err = converter{}.ConvertValue(nv.Value)	4,110✔
639	return	4,110✔
640	}	4,110✔
641
642	// ResetSession implements driver.SessionResetter.
643	// (From Go 1.10)
644	func (mc *mysqlConn) ResetSession(ctx context.Context) error {	13,900✔
645	if mc.closed.Load() {	13,900✔
646	return driver.ErrBadConn	×
647	}	×
648
649	// Perform a stale connection check. We only perform this check for
650	// the first query on a connection that has been checked out of the
651	// connection pool: a fresh connection from the pool is more likely
652	// to be stale, and it has not performed any previous writes that
653	// could cause data corruption, so it's safe to return ErrBadConn
654	// if the check fails.
655	if mc.cfg.CheckConnLiveness {	27,800✔
656	conn := mc.netConn	13,900✔
657	if mc.rawConn != nil {	14,200✔
658	conn = mc.rawConn	300✔
659	}	300✔
660	var err error	13,900✔
661	if mc.cfg.ReadTimeout != 0 {	13,900✔
662	err = conn.SetReadDeadline(time.Now().Add(mc.cfg.ReadTimeout))	×
663	}	×
664	if err == nil {	27,800✔
665	err = connCheck(conn)	13,900✔
666	}	13,900✔
667	if err != nil {	13,921✔
668	mc.cfg.Logger.Print("closing bad idle connection: ", err)	21✔
669	return driver.ErrBadConn	21✔
670	}	21✔
671	}
672
673	return nil	13,879✔
674	}
675
676	// IsValid implements driver.Validator interface
677	// (From Go 1.15)
678	func (mc *mysqlConn) IsValid() bool {	21,697✔
679	return !mc.closed.Load()	21,697✔
680	}	21,697✔

go-sql-driver / mysql / 7194530928

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous