mirror of
https://github.com/golang/go.git
synced 2025-05-18 13:54:40 +00:00
When context methods were initially added it was attempted to unify behavior between drivers without Context methods and those with Context methods to always return right away when the Context expired. However in doing so the driver call could be executed outside of the scope of the driver connection lock and thus bypassing thread safety. The new behavior waits until the driver operation is complete. It then checks to see if the context has expired and if so returns that error. Change-Id: I4a5c7c3263420c57778f36a5ed6fa0ef8cb32b20 Reviewed-on: https://go-review.googlesource.com/32422 Reviewed-by: Brad Fitzpatrick <bradfitz@golang.org> Run-TryBot: Brad Fitzpatrick <bradfitz@golang.org> TryBot-Result: Gobot Gobot <gobot@golang.org>
2431 lines
66 KiB
Go
2431 lines
66 KiB
Go
// Copyright 2011 The Go Authors. All rights reserved.
|
|
// Use of this source code is governed by a BSD-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
// Package sql provides a generic interface around SQL (or SQL-like)
|
|
// databases.
|
|
//
|
|
// The sql package must be used in conjunction with a database driver.
|
|
// See https://golang.org/s/sqldrivers for a list of drivers.
|
|
//
|
|
// For more usage examples, see the wiki page at
|
|
// https://golang.org/s/sqlwiki.
|
|
package sql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql/driver"
|
|
"database/sql/internal"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"reflect"
|
|
"runtime"
|
|
"sort"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
var (
|
|
driversMu sync.RWMutex
|
|
drivers = make(map[string]driver.Driver)
|
|
)
|
|
|
|
// nowFunc returns the current time; it's overridden in tests.
|
|
var nowFunc = time.Now
|
|
|
|
// Register makes a database driver available by the provided name.
|
|
// If Register is called twice with the same name or if driver is nil,
|
|
// it panics.
|
|
func Register(name string, driver driver.Driver) {
|
|
driversMu.Lock()
|
|
defer driversMu.Unlock()
|
|
if driver == nil {
|
|
panic("sql: Register driver is nil")
|
|
}
|
|
if _, dup := drivers[name]; dup {
|
|
panic("sql: Register called twice for driver " + name)
|
|
}
|
|
drivers[name] = driver
|
|
}
|
|
|
|
func unregisterAllDrivers() {
|
|
driversMu.Lock()
|
|
defer driversMu.Unlock()
|
|
// For tests.
|
|
drivers = make(map[string]driver.Driver)
|
|
}
|
|
|
|
// Drivers returns a sorted list of the names of the registered drivers.
|
|
func Drivers() []string {
|
|
driversMu.RLock()
|
|
defer driversMu.RUnlock()
|
|
var list []string
|
|
for name := range drivers {
|
|
list = append(list, name)
|
|
}
|
|
sort.Strings(list)
|
|
return list
|
|
}
|
|
|
|
// NamedParam may be passed into query parameter arguments to associate
|
|
// a named placeholder with a value.
|
|
type NamedParam struct {
|
|
_Named_Fields_Required struct{}
|
|
|
|
// Name of the parameter placeholder. If empty the ordinal position in the
|
|
// argument list will be used.
|
|
Name string
|
|
|
|
// Value of the parameter. It may be assigned the same value types as
|
|
// the query arguments.
|
|
Value interface{}
|
|
}
|
|
|
|
// Param provides a more concise way to create NamedParam values.
|
|
func Param(name string, value interface{}) NamedParam {
|
|
// This method exists because the go1compat promise
|
|
// doesn't guarantee that structs don't grow more fields,
|
|
// so unkeyed struct literals are a vet error. Thus, we don't
|
|
// want to encourage sql.NamedParam{name, value}.
|
|
return NamedParam{Name: name, Value: value}
|
|
}
|
|
|
|
// IsolationLevel is the transaction isolation level stored in Context.
|
|
// The IsolationLevel is set with IsolationContext and the context
|
|
// should be passed to BeginContext.
|
|
type IsolationLevel int
|
|
|
|
// Various isolation levels that drivers may support in BeginContext.
|
|
// If a driver does not support a given isolation level an error may be returned.
|
|
//
|
|
// See https://en.wikipedia.org/wiki/Isolation_(database_systems)#Isolation_levels.
|
|
const (
|
|
LevelDefault IsolationLevel = iota
|
|
LevelReadUncommitted
|
|
LevelReadCommitted
|
|
LevelWriteCommitted
|
|
LevelRepeatableRead
|
|
LevelSnapshot
|
|
LevelSerializable
|
|
LevelLinearizable
|
|
)
|
|
|
|
// IsolationContext returns a new Context that carries the provided isolation level.
|
|
// The context must contain the isolation level before beginning the transaction
|
|
// with BeginContext.
|
|
func IsolationContext(ctx context.Context, level IsolationLevel) context.Context {
|
|
return context.WithValue(ctx, internal.IsolationLevelKey{}, driver.IsolationLevel(level))
|
|
}
|
|
|
|
// ReadOnlyWithContext returns a new Context that carries the provided
|
|
// read-only transaction property. The context must contain the read-only property
|
|
// before beginning the transaction with BeginContext.
|
|
func ReadOnlyContext(ctx context.Context) context.Context {
|
|
return context.WithValue(ctx, internal.ReadOnlyKey{}, true)
|
|
}
|
|
|
|
// RawBytes is a byte slice that holds a reference to memory owned by
|
|
// the database itself. After a Scan into a RawBytes, the slice is only
|
|
// valid until the next call to Next, Scan, or Close.
|
|
type RawBytes []byte
|
|
|
|
// NullString represents a string that may be null.
|
|
// NullString implements the Scanner interface so
|
|
// it can be used as a scan destination:
|
|
//
|
|
// var s NullString
|
|
// err := db.QueryRow("SELECT name FROM foo WHERE id=?", id).Scan(&s)
|
|
// ...
|
|
// if s.Valid {
|
|
// // use s.String
|
|
// } else {
|
|
// // NULL value
|
|
// }
|
|
//
|
|
type NullString struct {
|
|
String string
|
|
Valid bool // Valid is true if String is not NULL
|
|
}
|
|
|
|
// Scan implements the Scanner interface.
|
|
func (ns *NullString) Scan(value interface{}) error {
|
|
if value == nil {
|
|
ns.String, ns.Valid = "", false
|
|
return nil
|
|
}
|
|
ns.Valid = true
|
|
return convertAssign(&ns.String, value)
|
|
}
|
|
|
|
// Value implements the driver Valuer interface.
|
|
func (ns NullString) Value() (driver.Value, error) {
|
|
if !ns.Valid {
|
|
return nil, nil
|
|
}
|
|
return ns.String, nil
|
|
}
|
|
|
|
// NullInt64 represents an int64 that may be null.
|
|
// NullInt64 implements the Scanner interface so
|
|
// it can be used as a scan destination, similar to NullString.
|
|
type NullInt64 struct {
|
|
Int64 int64
|
|
Valid bool // Valid is true if Int64 is not NULL
|
|
}
|
|
|
|
// Scan implements the Scanner interface.
|
|
func (n *NullInt64) Scan(value interface{}) error {
|
|
if value == nil {
|
|
n.Int64, n.Valid = 0, false
|
|
return nil
|
|
}
|
|
n.Valid = true
|
|
return convertAssign(&n.Int64, value)
|
|
}
|
|
|
|
// Value implements the driver Valuer interface.
|
|
func (n NullInt64) Value() (driver.Value, error) {
|
|
if !n.Valid {
|
|
return nil, nil
|
|
}
|
|
return n.Int64, nil
|
|
}
|
|
|
|
// NullFloat64 represents a float64 that may be null.
|
|
// NullFloat64 implements the Scanner interface so
|
|
// it can be used as a scan destination, similar to NullString.
|
|
type NullFloat64 struct {
|
|
Float64 float64
|
|
Valid bool // Valid is true if Float64 is not NULL
|
|
}
|
|
|
|
// Scan implements the Scanner interface.
|
|
func (n *NullFloat64) Scan(value interface{}) error {
|
|
if value == nil {
|
|
n.Float64, n.Valid = 0, false
|
|
return nil
|
|
}
|
|
n.Valid = true
|
|
return convertAssign(&n.Float64, value)
|
|
}
|
|
|
|
// Value implements the driver Valuer interface.
|
|
func (n NullFloat64) Value() (driver.Value, error) {
|
|
if !n.Valid {
|
|
return nil, nil
|
|
}
|
|
return n.Float64, nil
|
|
}
|
|
|
|
// NullBool represents a bool that may be null.
|
|
// NullBool implements the Scanner interface so
|
|
// it can be used as a scan destination, similar to NullString.
|
|
type NullBool struct {
|
|
Bool bool
|
|
Valid bool // Valid is true if Bool is not NULL
|
|
}
|
|
|
|
// Scan implements the Scanner interface.
|
|
func (n *NullBool) Scan(value interface{}) error {
|
|
if value == nil {
|
|
n.Bool, n.Valid = false, false
|
|
return nil
|
|
}
|
|
n.Valid = true
|
|
return convertAssign(&n.Bool, value)
|
|
}
|
|
|
|
// Value implements the driver Valuer interface.
|
|
func (n NullBool) Value() (driver.Value, error) {
|
|
if !n.Valid {
|
|
return nil, nil
|
|
}
|
|
return n.Bool, nil
|
|
}
|
|
|
|
// Scanner is an interface used by Scan.
|
|
type Scanner interface {
|
|
// Scan assigns a value from a database driver.
|
|
//
|
|
// The src value will be of one of the following types:
|
|
//
|
|
// int64
|
|
// float64
|
|
// bool
|
|
// []byte
|
|
// string
|
|
// time.Time
|
|
// nil - for NULL values
|
|
//
|
|
// An error should be returned if the value cannot be stored
|
|
// without loss of information.
|
|
Scan(src interface{}) error
|
|
}
|
|
|
|
// ErrNoRows is returned by Scan when QueryRow doesn't return a
|
|
// row. In such a case, QueryRow returns a placeholder *Row value that
|
|
// defers this error until a Scan.
|
|
var ErrNoRows = errors.New("sql: no rows in result set")
|
|
|
|
// DB is a database handle representing a pool of zero or more
|
|
// underlying connections. It's safe for concurrent use by multiple
|
|
// goroutines.
|
|
//
|
|
// The sql package creates and frees connections automatically; it
|
|
// also maintains a free pool of idle connections. If the database has
|
|
// a concept of per-connection state, such state can only be reliably
|
|
// observed within a transaction. Once DB.Begin is called, the
|
|
// returned Tx is bound to a single connection. Once Commit or
|
|
// Rollback is called on the transaction, that transaction's
|
|
// connection is returned to DB's idle connection pool. The pool size
|
|
// can be controlled with SetMaxIdleConns.
|
|
type DB struct {
|
|
driver driver.Driver
|
|
dsn string
|
|
// numClosed is an atomic counter which represents a total number of
|
|
// closed connections. Stmt.openStmt checks it before cleaning closed
|
|
// connections in Stmt.css.
|
|
numClosed uint64
|
|
|
|
mu sync.Mutex // protects following fields
|
|
freeConn []*driverConn
|
|
connRequests []chan connRequest
|
|
numOpen int // number of opened and pending open connections
|
|
// Used to signal the need for new connections
|
|
// a goroutine running connectionOpener() reads on this chan and
|
|
// maybeOpenNewConnections sends on the chan (one send per needed connection)
|
|
// It is closed during db.Close(). The close tells the connectionOpener
|
|
// goroutine to exit.
|
|
openerCh chan struct{}
|
|
closed bool
|
|
dep map[finalCloser]depSet
|
|
lastPut map[*driverConn]string // stacktrace of last conn's put; debug only
|
|
maxIdle int // zero means defaultMaxIdleConns; negative means 0
|
|
maxOpen int // <= 0 means unlimited
|
|
maxLifetime time.Duration // maximum amount of time a connection may be reused
|
|
cleanerCh chan struct{}
|
|
}
|
|
|
|
// connReuseStrategy determines how (*DB).conn returns database connections.
|
|
type connReuseStrategy uint8
|
|
|
|
const (
|
|
// alwaysNewConn forces a new connection to the database.
|
|
alwaysNewConn connReuseStrategy = iota
|
|
// cachedOrNewConn returns a cached connection, if available, else waits
|
|
// for one to become available (if MaxOpenConns has been reached) or
|
|
// creates a new database connection.
|
|
cachedOrNewConn
|
|
)
|
|
|
|
// driverConn wraps a driver.Conn with a mutex, to
|
|
// be held during all calls into the Conn. (including any calls onto
|
|
// interfaces returned via that Conn, such as calls on Tx, Stmt,
|
|
// Result, Rows)
|
|
type driverConn struct {
|
|
db *DB
|
|
createdAt time.Time
|
|
|
|
sync.Mutex // guards following
|
|
ci driver.Conn
|
|
closed bool
|
|
finalClosed bool // ci.Close has been called
|
|
openStmt map[*driverStmt]bool
|
|
|
|
// guarded by db.mu
|
|
inUse bool
|
|
onPut []func() // code (with db.mu held) run when conn is next returned
|
|
dbmuClosed bool // same as closed, but guarded by db.mu, for removeClosedStmtLocked
|
|
}
|
|
|
|
func (dc *driverConn) releaseConn(err error) {
|
|
dc.db.putConn(dc, err)
|
|
}
|
|
|
|
func (dc *driverConn) removeOpenStmt(ds *driverStmt) {
|
|
dc.Lock()
|
|
defer dc.Unlock()
|
|
delete(dc.openStmt, ds)
|
|
}
|
|
|
|
func (dc *driverConn) expired(timeout time.Duration) bool {
|
|
if timeout <= 0 {
|
|
return false
|
|
}
|
|
return dc.createdAt.Add(timeout).Before(nowFunc())
|
|
}
|
|
|
|
func (dc *driverConn) prepareLocked(ctx context.Context, query string) (*driverStmt, error) {
|
|
si, err := ctxDriverPrepare(ctx, dc.ci, query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Track each driverConn's open statements, so we can close them
|
|
// before closing the conn.
|
|
//
|
|
// Wrap all driver.Stmt is *driverStmt to ensure they are only closed once.
|
|
if dc.openStmt == nil {
|
|
dc.openStmt = make(map[*driverStmt]bool)
|
|
}
|
|
ds := &driverStmt{Locker: dc, si: si}
|
|
dc.openStmt[ds] = true
|
|
|
|
return ds, nil
|
|
}
|
|
|
|
// the dc.db's Mutex is held.
|
|
func (dc *driverConn) closeDBLocked() func() error {
|
|
dc.Lock()
|
|
defer dc.Unlock()
|
|
if dc.closed {
|
|
return func() error { return errors.New("sql: duplicate driverConn close") }
|
|
}
|
|
dc.closed = true
|
|
return dc.db.removeDepLocked(dc, dc)
|
|
}
|
|
|
|
func (dc *driverConn) Close() error {
|
|
dc.Lock()
|
|
if dc.closed {
|
|
dc.Unlock()
|
|
return errors.New("sql: duplicate driverConn close")
|
|
}
|
|
dc.closed = true
|
|
dc.Unlock() // not defer; removeDep finalClose calls may need to lock
|
|
|
|
// And now updates that require holding dc.mu.Lock.
|
|
dc.db.mu.Lock()
|
|
dc.dbmuClosed = true
|
|
fn := dc.db.removeDepLocked(dc, dc)
|
|
dc.db.mu.Unlock()
|
|
return fn()
|
|
}
|
|
|
|
func (dc *driverConn) finalClose() error {
|
|
var err error
|
|
|
|
// Each *driverStmt has a lock to the dc. Copy the list out of the dc
|
|
// before calling close on each stmt.
|
|
var openStmt []*driverStmt
|
|
withLock(dc, func() {
|
|
openStmt = make([]*driverStmt, 0, len(dc.openStmt))
|
|
for ds := range dc.openStmt {
|
|
openStmt = append(openStmt, ds)
|
|
}
|
|
dc.openStmt = nil
|
|
})
|
|
for _, ds := range openStmt {
|
|
ds.Close()
|
|
}
|
|
withLock(dc, func() {
|
|
dc.finalClosed = true
|
|
err = dc.ci.Close()
|
|
dc.ci = nil
|
|
})
|
|
|
|
dc.db.mu.Lock()
|
|
dc.db.numOpen--
|
|
dc.db.maybeOpenNewConnections()
|
|
dc.db.mu.Unlock()
|
|
|
|
atomic.AddUint64(&dc.db.numClosed, 1)
|
|
return err
|
|
}
|
|
|
|
// driverStmt associates a driver.Stmt with the
|
|
// *driverConn from which it came, so the driverConn's lock can be
|
|
// held during calls.
|
|
type driverStmt struct {
|
|
sync.Locker // the *driverConn
|
|
si driver.Stmt
|
|
closed bool
|
|
closeErr error // return value of previous Close call
|
|
}
|
|
|
|
// Close ensures dirver.Stmt is only closed once any always returns the same
|
|
// result.
|
|
func (ds *driverStmt) Close() error {
|
|
ds.Lock()
|
|
defer ds.Unlock()
|
|
if ds.closed {
|
|
return ds.closeErr
|
|
}
|
|
ds.closed = true
|
|
ds.closeErr = ds.si.Close()
|
|
return ds.closeErr
|
|
}
|
|
|
|
// depSet is a finalCloser's outstanding dependencies
|
|
type depSet map[interface{}]bool // set of true bools
|
|
|
|
// The finalCloser interface is used by (*DB).addDep and related
|
|
// dependency reference counting.
|
|
type finalCloser interface {
|
|
// finalClose is called when the reference count of an object
|
|
// goes to zero. (*DB).mu is not held while calling it.
|
|
finalClose() error
|
|
}
|
|
|
|
// addDep notes that x now depends on dep, and x's finalClose won't be
|
|
// called until all of x's dependencies are removed with removeDep.
|
|
func (db *DB) addDep(x finalCloser, dep interface{}) {
|
|
//println(fmt.Sprintf("addDep(%T %p, %T %p)", x, x, dep, dep))
|
|
db.mu.Lock()
|
|
defer db.mu.Unlock()
|
|
db.addDepLocked(x, dep)
|
|
}
|
|
|
|
func (db *DB) addDepLocked(x finalCloser, dep interface{}) {
|
|
if db.dep == nil {
|
|
db.dep = make(map[finalCloser]depSet)
|
|
}
|
|
xdep := db.dep[x]
|
|
if xdep == nil {
|
|
xdep = make(depSet)
|
|
db.dep[x] = xdep
|
|
}
|
|
xdep[dep] = true
|
|
}
|
|
|
|
// removeDep notes that x no longer depends on dep.
|
|
// If x still has dependencies, nil is returned.
|
|
// If x no longer has any dependencies, its finalClose method will be
|
|
// called and its error value will be returned.
|
|
func (db *DB) removeDep(x finalCloser, dep interface{}) error {
|
|
db.mu.Lock()
|
|
fn := db.removeDepLocked(x, dep)
|
|
db.mu.Unlock()
|
|
return fn()
|
|
}
|
|
|
|
func (db *DB) removeDepLocked(x finalCloser, dep interface{}) func() error {
|
|
//println(fmt.Sprintf("removeDep(%T %p, %T %p)", x, x, dep, dep))
|
|
|
|
xdep, ok := db.dep[x]
|
|
if !ok {
|
|
panic(fmt.Sprintf("unpaired removeDep: no deps for %T", x))
|
|
}
|
|
|
|
l0 := len(xdep)
|
|
delete(xdep, dep)
|
|
|
|
switch len(xdep) {
|
|
case l0:
|
|
// Nothing removed. Shouldn't happen.
|
|
panic(fmt.Sprintf("unpaired removeDep: no %T dep on %T", dep, x))
|
|
case 0:
|
|
// No more dependencies.
|
|
delete(db.dep, x)
|
|
return x.finalClose
|
|
default:
|
|
// Dependencies remain.
|
|
return func() error { return nil }
|
|
}
|
|
}
|
|
|
|
// This is the size of the connectionOpener request chan (DB.openerCh).
|
|
// This value should be larger than the maximum typical value
|
|
// used for db.maxOpen. If maxOpen is significantly larger than
|
|
// connectionRequestQueueSize then it is possible for ALL calls into the *DB
|
|
// to block until the connectionOpener can satisfy the backlog of requests.
|
|
var connectionRequestQueueSize = 1000000
|
|
|
|
// Open opens a database specified by its database driver name and a
|
|
// driver-specific data source name, usually consisting of at least a
|
|
// database name and connection information.
|
|
//
|
|
// Most users will open a database via a driver-specific connection
|
|
// helper function that returns a *DB. No database drivers are included
|
|
// in the Go standard library. See https://golang.org/s/sqldrivers for
|
|
// a list of third-party drivers.
|
|
//
|
|
// Open may just validate its arguments without creating a connection
|
|
// to the database. To verify that the data source name is valid, call
|
|
// Ping.
|
|
//
|
|
// The returned DB is safe for concurrent use by multiple goroutines
|
|
// and maintains its own pool of idle connections. Thus, the Open
|
|
// function should be called just once. It is rarely necessary to
|
|
// close a DB.
|
|
func Open(driverName, dataSourceName string) (*DB, error) {
|
|
driversMu.RLock()
|
|
driveri, ok := drivers[driverName]
|
|
driversMu.RUnlock()
|
|
if !ok {
|
|
return nil, fmt.Errorf("sql: unknown driver %q (forgotten import?)", driverName)
|
|
}
|
|
db := &DB{
|
|
driver: driveri,
|
|
dsn: dataSourceName,
|
|
openerCh: make(chan struct{}, connectionRequestQueueSize),
|
|
lastPut: make(map[*driverConn]string),
|
|
}
|
|
go db.connectionOpener()
|
|
return db, nil
|
|
}
|
|
|
|
// PingContext verifies a connection to the database is still alive,
|
|
// establishing a connection if necessary.
|
|
func (db *DB) PingContext(ctx context.Context) error {
|
|
var dc *driverConn
|
|
var err error
|
|
|
|
for i := 0; i < maxBadConnRetries; i++ {
|
|
dc, err = db.conn(ctx, cachedOrNewConn)
|
|
if err != driver.ErrBadConn {
|
|
break
|
|
}
|
|
}
|
|
if err == driver.ErrBadConn {
|
|
dc, err = db.conn(ctx, alwaysNewConn)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if pinger, ok := dc.ci.(driver.Pinger); ok {
|
|
err = pinger.Ping(ctx)
|
|
}
|
|
db.putConn(dc, err)
|
|
return err
|
|
}
|
|
|
|
// Ping verifies a connection to the database is still alive,
|
|
// establishing a connection if necessary.
|
|
func (db *DB) Ping() error {
|
|
return db.PingContext(context.Background())
|
|
}
|
|
|
|
// Close closes the database, releasing any open resources.
|
|
//
|
|
// It is rare to Close a DB, as the DB handle is meant to be
|
|
// long-lived and shared between many goroutines.
|
|
func (db *DB) Close() error {
|
|
db.mu.Lock()
|
|
if db.closed { // Make DB.Close idempotent
|
|
db.mu.Unlock()
|
|
return nil
|
|
}
|
|
close(db.openerCh)
|
|
if db.cleanerCh != nil {
|
|
close(db.cleanerCh)
|
|
}
|
|
var err error
|
|
fns := make([]func() error, 0, len(db.freeConn))
|
|
for _, dc := range db.freeConn {
|
|
fns = append(fns, dc.closeDBLocked())
|
|
}
|
|
db.freeConn = nil
|
|
db.closed = true
|
|
for _, req := range db.connRequests {
|
|
close(req)
|
|
}
|
|
db.mu.Unlock()
|
|
for _, fn := range fns {
|
|
err1 := fn()
|
|
if err1 != nil {
|
|
err = err1
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
const defaultMaxIdleConns = 2
|
|
|
|
func (db *DB) maxIdleConnsLocked() int {
|
|
n := db.maxIdle
|
|
switch {
|
|
case n == 0:
|
|
// TODO(bradfitz): ask driver, if supported, for its default preference
|
|
return defaultMaxIdleConns
|
|
case n < 0:
|
|
return 0
|
|
default:
|
|
return n
|
|
}
|
|
}
|
|
|
|
// SetMaxIdleConns sets the maximum number of connections in the idle
|
|
// connection pool.
|
|
//
|
|
// If MaxOpenConns is greater than 0 but less than the new MaxIdleConns
|
|
// then the new MaxIdleConns will be reduced to match the MaxOpenConns limit
|
|
//
|
|
// If n <= 0, no idle connections are retained.
|
|
func (db *DB) SetMaxIdleConns(n int) {
|
|
db.mu.Lock()
|
|
if n > 0 {
|
|
db.maxIdle = n
|
|
} else {
|
|
// No idle connections.
|
|
db.maxIdle = -1
|
|
}
|
|
// Make sure maxIdle doesn't exceed maxOpen
|
|
if db.maxOpen > 0 && db.maxIdleConnsLocked() > db.maxOpen {
|
|
db.maxIdle = db.maxOpen
|
|
}
|
|
var closing []*driverConn
|
|
idleCount := len(db.freeConn)
|
|
maxIdle := db.maxIdleConnsLocked()
|
|
if idleCount > maxIdle {
|
|
closing = db.freeConn[maxIdle:]
|
|
db.freeConn = db.freeConn[:maxIdle]
|
|
}
|
|
db.mu.Unlock()
|
|
for _, c := range closing {
|
|
c.Close()
|
|
}
|
|
}
|
|
|
|
// SetMaxOpenConns sets the maximum number of open connections to the database.
|
|
//
|
|
// If MaxIdleConns is greater than 0 and the new MaxOpenConns is less than
|
|
// MaxIdleConns, then MaxIdleConns will be reduced to match the new
|
|
// MaxOpenConns limit
|
|
//
|
|
// If n <= 0, then there is no limit on the number of open connections.
|
|
// The default is 0 (unlimited).
|
|
func (db *DB) SetMaxOpenConns(n int) {
|
|
db.mu.Lock()
|
|
db.maxOpen = n
|
|
if n < 0 {
|
|
db.maxOpen = 0
|
|
}
|
|
syncMaxIdle := db.maxOpen > 0 && db.maxIdleConnsLocked() > db.maxOpen
|
|
db.mu.Unlock()
|
|
if syncMaxIdle {
|
|
db.SetMaxIdleConns(n)
|
|
}
|
|
}
|
|
|
|
// SetConnMaxLifetime sets the maximum amount of time a connection may be reused.
|
|
//
|
|
// Expired connections may be closed lazily before reuse.
|
|
//
|
|
// If d <= 0, connections are reused forever.
|
|
func (db *DB) SetConnMaxLifetime(d time.Duration) {
|
|
if d < 0 {
|
|
d = 0
|
|
}
|
|
db.mu.Lock()
|
|
// wake cleaner up when lifetime is shortened.
|
|
if d > 0 && d < db.maxLifetime && db.cleanerCh != nil {
|
|
select {
|
|
case db.cleanerCh <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
db.maxLifetime = d
|
|
db.startCleanerLocked()
|
|
db.mu.Unlock()
|
|
}
|
|
|
|
// startCleanerLocked starts connectionCleaner if needed.
|
|
func (db *DB) startCleanerLocked() {
|
|
if db.maxLifetime > 0 && db.numOpen > 0 && db.cleanerCh == nil {
|
|
db.cleanerCh = make(chan struct{}, 1)
|
|
go db.connectionCleaner(db.maxLifetime)
|
|
}
|
|
}
|
|
|
|
func (db *DB) connectionCleaner(d time.Duration) {
|
|
const minInterval = time.Second
|
|
|
|
if d < minInterval {
|
|
d = minInterval
|
|
}
|
|
t := time.NewTimer(d)
|
|
|
|
for {
|
|
select {
|
|
case <-t.C:
|
|
case <-db.cleanerCh: // maxLifetime was changed or db was closed.
|
|
}
|
|
|
|
db.mu.Lock()
|
|
d = db.maxLifetime
|
|
if db.closed || db.numOpen == 0 || d <= 0 {
|
|
db.cleanerCh = nil
|
|
db.mu.Unlock()
|
|
return
|
|
}
|
|
|
|
expiredSince := nowFunc().Add(-d)
|
|
var closing []*driverConn
|
|
for i := 0; i < len(db.freeConn); i++ {
|
|
c := db.freeConn[i]
|
|
if c.createdAt.Before(expiredSince) {
|
|
closing = append(closing, c)
|
|
last := len(db.freeConn) - 1
|
|
db.freeConn[i] = db.freeConn[last]
|
|
db.freeConn[last] = nil
|
|
db.freeConn = db.freeConn[:last]
|
|
i--
|
|
}
|
|
}
|
|
db.mu.Unlock()
|
|
|
|
for _, c := range closing {
|
|
c.Close()
|
|
}
|
|
|
|
if d < minInterval {
|
|
d = minInterval
|
|
}
|
|
t.Reset(d)
|
|
}
|
|
}
|
|
|
|
// DBStats contains database statistics.
|
|
type DBStats struct {
|
|
// OpenConnections is the number of open connections to the database.
|
|
OpenConnections int
|
|
}
|
|
|
|
// Stats returns database statistics.
|
|
func (db *DB) Stats() DBStats {
|
|
db.mu.Lock()
|
|
stats := DBStats{
|
|
OpenConnections: db.numOpen,
|
|
}
|
|
db.mu.Unlock()
|
|
return stats
|
|
}
|
|
|
|
// Assumes db.mu is locked.
|
|
// If there are connRequests and the connection limit hasn't been reached,
|
|
// then tell the connectionOpener to open new connections.
|
|
func (db *DB) maybeOpenNewConnections() {
|
|
numRequests := len(db.connRequests)
|
|
if db.maxOpen > 0 {
|
|
numCanOpen := db.maxOpen - db.numOpen
|
|
if numRequests > numCanOpen {
|
|
numRequests = numCanOpen
|
|
}
|
|
}
|
|
for numRequests > 0 {
|
|
db.numOpen++ // optimistically
|
|
numRequests--
|
|
if db.closed {
|
|
return
|
|
}
|
|
db.openerCh <- struct{}{}
|
|
}
|
|
}
|
|
|
|
// Runs in a separate goroutine, opens new connections when requested.
|
|
func (db *DB) connectionOpener() {
|
|
for range db.openerCh {
|
|
db.openNewConnection()
|
|
}
|
|
}
|
|
|
|
// Open one new connection
|
|
func (db *DB) openNewConnection() {
|
|
// maybeOpenNewConnctions has already executed db.numOpen++ before it sent
|
|
// on db.openerCh. This function must execute db.numOpen-- if the
|
|
// connection fails or is closed before returning.
|
|
ci, err := db.driver.Open(db.dsn)
|
|
db.mu.Lock()
|
|
defer db.mu.Unlock()
|
|
if db.closed {
|
|
if err == nil {
|
|
ci.Close()
|
|
}
|
|
db.numOpen--
|
|
return
|
|
}
|
|
if err != nil {
|
|
db.numOpen--
|
|
db.putConnDBLocked(nil, err)
|
|
db.maybeOpenNewConnections()
|
|
return
|
|
}
|
|
dc := &driverConn{
|
|
db: db,
|
|
createdAt: nowFunc(),
|
|
ci: ci,
|
|
}
|
|
if db.putConnDBLocked(dc, err) {
|
|
db.addDepLocked(dc, dc)
|
|
} else {
|
|
db.numOpen--
|
|
ci.Close()
|
|
}
|
|
}
|
|
|
|
// connRequest represents one request for a new connection
|
|
// When there are no idle connections available, DB.conn will create
|
|
// a new connRequest and put it on the db.connRequests list.
|
|
type connRequest struct {
|
|
conn *driverConn
|
|
err error
|
|
}
|
|
|
|
var errDBClosed = errors.New("sql: database is closed")
|
|
|
|
// conn returns a newly-opened or cached *driverConn.
|
|
func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) {
|
|
db.mu.Lock()
|
|
if db.closed {
|
|
db.mu.Unlock()
|
|
return nil, errDBClosed
|
|
}
|
|
// Check if the context is expired.
|
|
select {
|
|
default:
|
|
case <-ctx.Done():
|
|
db.mu.Unlock()
|
|
return nil, ctx.Err()
|
|
}
|
|
lifetime := db.maxLifetime
|
|
|
|
// Prefer a free connection, if possible.
|
|
numFree := len(db.freeConn)
|
|
if strategy == cachedOrNewConn && numFree > 0 {
|
|
conn := db.freeConn[0]
|
|
copy(db.freeConn, db.freeConn[1:])
|
|
db.freeConn = db.freeConn[:numFree-1]
|
|
conn.inUse = true
|
|
db.mu.Unlock()
|
|
if conn.expired(lifetime) {
|
|
conn.Close()
|
|
return nil, driver.ErrBadConn
|
|
}
|
|
return conn, nil
|
|
}
|
|
|
|
// Out of free connections or we were asked not to use one. If we're not
|
|
// allowed to open any more connections, make a request and wait.
|
|
if db.maxOpen > 0 && db.numOpen >= db.maxOpen {
|
|
// Make the connRequest channel. It's buffered so that the
|
|
// connectionOpener doesn't block while waiting for the req to be read.
|
|
req := make(chan connRequest, 1)
|
|
db.connRequests = append(db.connRequests, req)
|
|
db.mu.Unlock()
|
|
|
|
// Timeout the connection request with the context.
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case ret, ok := <-req:
|
|
if !ok {
|
|
return nil, errDBClosed
|
|
}
|
|
if ret.err == nil && ret.conn.expired(lifetime) {
|
|
ret.conn.Close()
|
|
return nil, driver.ErrBadConn
|
|
}
|
|
return ret.conn, ret.err
|
|
}
|
|
}
|
|
|
|
db.numOpen++ // optimistically
|
|
db.mu.Unlock()
|
|
ci, err := db.driver.Open(db.dsn)
|
|
if err != nil {
|
|
db.mu.Lock()
|
|
db.numOpen-- // correct for earlier optimism
|
|
db.maybeOpenNewConnections()
|
|
db.mu.Unlock()
|
|
return nil, err
|
|
}
|
|
db.mu.Lock()
|
|
dc := &driverConn{
|
|
db: db,
|
|
createdAt: nowFunc(),
|
|
ci: ci,
|
|
}
|
|
db.addDepLocked(dc, dc)
|
|
dc.inUse = true
|
|
db.mu.Unlock()
|
|
return dc, nil
|
|
}
|
|
|
|
// putConnHook is a hook for testing.
|
|
var putConnHook func(*DB, *driverConn)
|
|
|
|
// noteUnusedDriverStatement notes that ds is no longer used and should
|
|
// be closed whenever possible (when c is next not in use), unless c is
|
|
// already closed.
|
|
func (db *DB) noteUnusedDriverStatement(c *driverConn, ds *driverStmt) {
|
|
db.mu.Lock()
|
|
defer db.mu.Unlock()
|
|
if c.inUse {
|
|
c.onPut = append(c.onPut, func() {
|
|
ds.Close()
|
|
})
|
|
} else {
|
|
c.Lock()
|
|
fc := c.finalClosed
|
|
c.Unlock()
|
|
if !fc {
|
|
ds.Close()
|
|
}
|
|
}
|
|
}
|
|
|
|
// debugGetPut determines whether getConn & putConn calls' stack traces
|
|
// are returned for more verbose crashes.
|
|
const debugGetPut = false
|
|
|
|
// putConn adds a connection to the db's free pool.
|
|
// err is optionally the last error that occurred on this connection.
|
|
func (db *DB) putConn(dc *driverConn, err error) {
|
|
db.mu.Lock()
|
|
if !dc.inUse {
|
|
if debugGetPut {
|
|
fmt.Printf("putConn(%v) DUPLICATE was: %s\n\nPREVIOUS was: %s", dc, stack(), db.lastPut[dc])
|
|
}
|
|
panic("sql: connection returned that was never out")
|
|
}
|
|
if debugGetPut {
|
|
db.lastPut[dc] = stack()
|
|
}
|
|
dc.inUse = false
|
|
|
|
for _, fn := range dc.onPut {
|
|
fn()
|
|
}
|
|
dc.onPut = nil
|
|
|
|
if err == driver.ErrBadConn {
|
|
// Don't reuse bad connections.
|
|
// Since the conn is considered bad and is being discarded, treat it
|
|
// as closed. Don't decrement the open count here, finalClose will
|
|
// take care of that.
|
|
db.maybeOpenNewConnections()
|
|
db.mu.Unlock()
|
|
dc.Close()
|
|
return
|
|
}
|
|
if putConnHook != nil {
|
|
putConnHook(db, dc)
|
|
}
|
|
added := db.putConnDBLocked(dc, nil)
|
|
db.mu.Unlock()
|
|
|
|
if !added {
|
|
dc.Close()
|
|
}
|
|
}
|
|
|
|
// Satisfy a connRequest or put the driverConn in the idle pool and return true
|
|
// or return false.
|
|
// putConnDBLocked will satisfy a connRequest if there is one, or it will
|
|
// return the *driverConn to the freeConn list if err == nil and the idle
|
|
// connection limit will not be exceeded.
|
|
// If err != nil, the value of dc is ignored.
|
|
// If err == nil, then dc must not equal nil.
|
|
// If a connRequest was fulfilled or the *driverConn was placed in the
|
|
// freeConn list, then true is returned, otherwise false is returned.
|
|
func (db *DB) putConnDBLocked(dc *driverConn, err error) bool {
|
|
if db.closed {
|
|
return false
|
|
}
|
|
if db.maxOpen > 0 && db.numOpen > db.maxOpen {
|
|
return false
|
|
}
|
|
if c := len(db.connRequests); c > 0 {
|
|
req := db.connRequests[0]
|
|
// This copy is O(n) but in practice faster than a linked list.
|
|
// TODO: consider compacting it down less often and
|
|
// moving the base instead?
|
|
copy(db.connRequests, db.connRequests[1:])
|
|
db.connRequests = db.connRequests[:c-1]
|
|
if err == nil {
|
|
dc.inUse = true
|
|
}
|
|
req <- connRequest{
|
|
conn: dc,
|
|
err: err,
|
|
}
|
|
return true
|
|
} else if err == nil && !db.closed && db.maxIdleConnsLocked() > len(db.freeConn) {
|
|
db.freeConn = append(db.freeConn, dc)
|
|
db.startCleanerLocked()
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// maxBadConnRetries is the number of maximum retries if the driver returns
|
|
// driver.ErrBadConn to signal a broken connection before forcing a new
|
|
// connection to be opened.
|
|
const maxBadConnRetries = 2
|
|
|
|
// PrepareContext creates a prepared statement for later queries or executions.
|
|
// Multiple queries or executions may be run concurrently from the
|
|
// returned statement.
|
|
// The caller must call the statement's Close method
|
|
// when the statement is no longer needed.
|
|
//
|
|
// The provided context is used for the preparation of the statement, not for the
|
|
// execution of the statement.
|
|
func (db *DB) PrepareContext(ctx context.Context, query string) (*Stmt, error) {
|
|
var stmt *Stmt
|
|
var err error
|
|
for i := 0; i < maxBadConnRetries; i++ {
|
|
stmt, err = db.prepare(ctx, query, cachedOrNewConn)
|
|
if err != driver.ErrBadConn {
|
|
break
|
|
}
|
|
}
|
|
if err == driver.ErrBadConn {
|
|
return db.prepare(ctx, query, alwaysNewConn)
|
|
}
|
|
return stmt, err
|
|
}
|
|
|
|
// Prepare creates a prepared statement for later queries or executions.
|
|
// Multiple queries or executions may be run concurrently from the
|
|
// returned statement.
|
|
// The caller must call the statement's Close method
|
|
// when the statement is no longer needed.
|
|
func (db *DB) Prepare(query string) (*Stmt, error) {
|
|
return db.PrepareContext(context.Background(), query)
|
|
}
|
|
|
|
func (db *DB) prepare(ctx context.Context, query string, strategy connReuseStrategy) (*Stmt, error) {
|
|
// TODO: check if db.driver supports an optional
|
|
// driver.Preparer interface and call that instead, if so,
|
|
// otherwise we make a prepared statement that's bound
|
|
// to a connection, and to execute this prepared statement
|
|
// we either need to use this connection (if it's free), else
|
|
// get a new connection + re-prepare + execute on that one.
|
|
dc, err := db.conn(ctx, strategy)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var ds *driverStmt
|
|
withLock(dc, func() {
|
|
ds, err = dc.prepareLocked(ctx, query)
|
|
})
|
|
if err != nil {
|
|
db.putConn(dc, err)
|
|
return nil, err
|
|
}
|
|
stmt := &Stmt{
|
|
db: db,
|
|
query: query,
|
|
css: []connStmt{{dc, ds}},
|
|
lastNumClosed: atomic.LoadUint64(&db.numClosed),
|
|
}
|
|
db.addDep(stmt, stmt)
|
|
db.putConn(dc, nil)
|
|
return stmt, nil
|
|
}
|
|
|
|
// ExecContext executes a query without returning any rows.
|
|
// The args are for any placeholder parameters in the query.
|
|
func (db *DB) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error) {
|
|
var res Result
|
|
var err error
|
|
for i := 0; i < maxBadConnRetries; i++ {
|
|
res, err = db.exec(ctx, query, args, cachedOrNewConn)
|
|
if err != driver.ErrBadConn {
|
|
break
|
|
}
|
|
}
|
|
if err == driver.ErrBadConn {
|
|
return db.exec(ctx, query, args, alwaysNewConn)
|
|
}
|
|
return res, err
|
|
}
|
|
|
|
// Exec executes a query without returning any rows.
|
|
// The args are for any placeholder parameters in the query.
|
|
func (db *DB) Exec(query string, args ...interface{}) (Result, error) {
|
|
return db.ExecContext(context.Background(), query, args...)
|
|
}
|
|
|
|
func (db *DB) exec(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (res Result, err error) {
|
|
dc, err := db.conn(ctx, strategy)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() {
|
|
db.putConn(dc, err)
|
|
}()
|
|
|
|
if execer, ok := dc.ci.(driver.Execer); ok {
|
|
var dargs []driver.NamedValue
|
|
dargs, err = driverArgs(nil, args)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var resi driver.Result
|
|
withLock(dc, func() {
|
|
resi, err = ctxDriverExec(ctx, execer, query, dargs)
|
|
})
|
|
if err != driver.ErrSkip {
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return driverResult{dc, resi}, nil
|
|
}
|
|
}
|
|
|
|
var si driver.Stmt
|
|
withLock(dc, func() {
|
|
si, err = ctxDriverPrepare(ctx, dc.ci, query)
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ds := &driverStmt{Locker: dc, si: si}
|
|
defer ds.Close()
|
|
return resultFromStatement(ctx, ds, args...)
|
|
}
|
|
|
|
// QueryContext executes a query that returns rows, typically a SELECT.
|
|
// The args are for any placeholder parameters in the query.
|
|
func (db *DB) QueryContext(ctx context.Context, query string, args ...interface{}) (*Rows, error) {
|
|
var rows *Rows
|
|
var err error
|
|
for i := 0; i < maxBadConnRetries; i++ {
|
|
rows, err = db.query(ctx, query, args, cachedOrNewConn)
|
|
if err != driver.ErrBadConn {
|
|
break
|
|
}
|
|
}
|
|
if err == driver.ErrBadConn {
|
|
return db.query(ctx, query, args, alwaysNewConn)
|
|
}
|
|
return rows, err
|
|
}
|
|
|
|
// Query executes a query that returns rows, typically a SELECT.
|
|
// The args are for any placeholder parameters in the query.
|
|
func (db *DB) Query(query string, args ...interface{}) (*Rows, error) {
|
|
return db.QueryContext(context.Background(), query, args...)
|
|
}
|
|
|
|
func (db *DB) query(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (*Rows, error) {
|
|
ci, err := db.conn(ctx, strategy)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return db.queryConn(ctx, ci, ci.releaseConn, query, args)
|
|
}
|
|
|
|
// queryConn executes a query on the given connection.
|
|
// The connection gets released by the releaseConn function.
|
|
func (db *DB) queryConn(ctx context.Context, dc *driverConn, releaseConn func(error), query string, args []interface{}) (*Rows, error) {
|
|
if queryer, ok := dc.ci.(driver.Queryer); ok {
|
|
dargs, err := driverArgs(nil, args)
|
|
if err != nil {
|
|
releaseConn(err)
|
|
return nil, err
|
|
}
|
|
var rowsi driver.Rows
|
|
withLock(dc, func() {
|
|
rowsi, err = ctxDriverQuery(ctx, queryer, query, dargs)
|
|
})
|
|
if err != driver.ErrSkip {
|
|
if err != nil {
|
|
releaseConn(err)
|
|
return nil, err
|
|
}
|
|
// Note: ownership of dc passes to the *Rows, to be freed
|
|
// with releaseConn.
|
|
rows := &Rows{
|
|
dc: dc,
|
|
releaseConn: releaseConn,
|
|
rowsi: rowsi,
|
|
}
|
|
rows.initContextClose(ctx)
|
|
return rows, nil
|
|
}
|
|
}
|
|
|
|
var si driver.Stmt
|
|
var err error
|
|
withLock(dc, func() {
|
|
si, err = ctxDriverPrepare(ctx, dc.ci, query)
|
|
})
|
|
if err != nil {
|
|
releaseConn(err)
|
|
return nil, err
|
|
}
|
|
|
|
ds := &driverStmt{Locker: dc, si: si}
|
|
rowsi, err := rowsiFromStatement(ctx, ds, args...)
|
|
if err != nil {
|
|
ds.Close()
|
|
releaseConn(err)
|
|
return nil, err
|
|
}
|
|
|
|
// Note: ownership of ci passes to the *Rows, to be freed
|
|
// with releaseConn.
|
|
rows := &Rows{
|
|
dc: dc,
|
|
releaseConn: releaseConn,
|
|
rowsi: rowsi,
|
|
closeStmt: ds,
|
|
}
|
|
rows.initContextClose(ctx)
|
|
return rows, nil
|
|
}
|
|
|
|
// QueryRowContext executes a query that is expected to return at most one row.
|
|
// QueryRowContext always returns a non-nil value. Errors are deferred until
|
|
// Row's Scan method is called.
|
|
func (db *DB) QueryRowContext(ctx context.Context, query string, args ...interface{}) *Row {
|
|
rows, err := db.QueryContext(ctx, query, args...)
|
|
return &Row{rows: rows, err: err}
|
|
}
|
|
|
|
// QueryRow executes a query that is expected to return at most one row.
|
|
// QueryRow always returns a non-nil value. Errors are deferred until
|
|
// Row's Scan method is called.
|
|
func (db *DB) QueryRow(query string, args ...interface{}) *Row {
|
|
return db.QueryRowContext(context.Background(), query, args...)
|
|
}
|
|
|
|
// BeginContext starts a transaction.
|
|
//
|
|
// The provided context is used until the transaction is committed or rolled back.
|
|
// If the context is canceled, the sql package will roll back
|
|
// the transaction. Tx.Commit will return an error if the context provided to
|
|
// BeginContext is canceled.
|
|
//
|
|
// An isolation level may be set by setting the value in the context
|
|
// before calling this. If a non-default isolation level is used
|
|
// that the driver doesn't support an error will be returned. Different drivers
|
|
// may have slightly different meanings for the same isolation level.
|
|
func (db *DB) BeginContext(ctx context.Context) (*Tx, error) {
|
|
var tx *Tx
|
|
var err error
|
|
for i := 0; i < maxBadConnRetries; i++ {
|
|
tx, err = db.begin(ctx, cachedOrNewConn)
|
|
if err != driver.ErrBadConn {
|
|
break
|
|
}
|
|
}
|
|
if err == driver.ErrBadConn {
|
|
return db.begin(ctx, alwaysNewConn)
|
|
}
|
|
return tx, err
|
|
}
|
|
|
|
// Begin starts a transaction. The default isolation level is dependent on
|
|
// the driver.
|
|
func (db *DB) Begin() (*Tx, error) {
|
|
return db.BeginContext(context.Background())
|
|
}
|
|
|
|
func (db *DB) begin(ctx context.Context, strategy connReuseStrategy) (tx *Tx, err error) {
|
|
dc, err := db.conn(ctx, strategy)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var txi driver.Tx
|
|
withLock(dc, func() {
|
|
txi, err = ctxDriverBegin(ctx, dc.ci)
|
|
})
|
|
if err != nil {
|
|
db.putConn(dc, err)
|
|
return nil, err
|
|
}
|
|
|
|
// Schedule the transaction to rollback when the context is cancelled.
|
|
// The cancel function in Tx will be called after done is set to true.
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
tx = &Tx{
|
|
db: db,
|
|
dc: dc,
|
|
txi: txi,
|
|
cancel: cancel,
|
|
ctx: ctx,
|
|
}
|
|
go func(tx *Tx) {
|
|
select {
|
|
case <-tx.ctx.Done():
|
|
if !tx.isDone() {
|
|
// Discard and close the connection used to ensure the transaction
|
|
// is closed and the resources are released.
|
|
tx.rollback(true)
|
|
}
|
|
}
|
|
}(tx)
|
|
return tx, nil
|
|
}
|
|
|
|
// Driver returns the database's underlying driver.
|
|
func (db *DB) Driver() driver.Driver {
|
|
return db.driver
|
|
}
|
|
|
|
// Tx is an in-progress database transaction.
|
|
//
|
|
// A transaction must end with a call to Commit or Rollback.
|
|
//
|
|
// After a call to Commit or Rollback, all operations on the
|
|
// transaction fail with ErrTxDone.
|
|
//
|
|
// The statements prepared for a transaction by calling
|
|
// the transaction's Prepare or Stmt methods are closed
|
|
// by the call to Commit or Rollback.
|
|
type Tx struct {
|
|
db *DB
|
|
|
|
// dc is owned exclusively until Commit or Rollback, at which point
|
|
// it's returned with putConn.
|
|
dc *driverConn
|
|
txi driver.Tx
|
|
|
|
// done transitions from 0 to 1 exactly once, on Commit
|
|
// or Rollback. once done, all operations fail with
|
|
// ErrTxDone.
|
|
// Use atomic operations on value when checking value.
|
|
done int32
|
|
|
|
// All Stmts prepared for this transaction. These will be closed after the
|
|
// transaction has been committed or rolled back.
|
|
stmts struct {
|
|
sync.Mutex
|
|
v []*Stmt
|
|
}
|
|
|
|
// cancel is called after done transitions from false to true.
|
|
cancel func()
|
|
|
|
// ctx lives for the life of the transaction.
|
|
ctx context.Context
|
|
}
|
|
|
|
func (tx *Tx) isDone() bool {
|
|
return atomic.LoadInt32(&tx.done) != 0
|
|
}
|
|
|
|
// ErrTxDone is returned by any operation that is performed on a transaction
|
|
// that has already been committed or rolled back.
|
|
var ErrTxDone = errors.New("sql: Transaction has already been committed or rolled back")
|
|
|
|
func (tx *Tx) close(err error) {
|
|
if !atomic.CompareAndSwapInt32(&tx.done, 0, 1) {
|
|
panic("double close") // internal error
|
|
}
|
|
tx.db.putConn(tx.dc, err)
|
|
tx.cancel()
|
|
tx.dc = nil
|
|
tx.txi = nil
|
|
}
|
|
|
|
func (tx *Tx) grabConn(ctx context.Context) (*driverConn, error) {
|
|
if tx.isDone() {
|
|
return nil, ErrTxDone
|
|
}
|
|
return tx.dc, nil
|
|
}
|
|
|
|
// Closes all Stmts prepared for this transaction.
|
|
func (tx *Tx) closePrepared() {
|
|
tx.stmts.Lock()
|
|
defer tx.stmts.Unlock()
|
|
for _, stmt := range tx.stmts.v {
|
|
stmt.Close()
|
|
}
|
|
}
|
|
|
|
// Commit commits the transaction.
|
|
func (tx *Tx) Commit() error {
|
|
select {
|
|
default:
|
|
case <-tx.ctx.Done():
|
|
return tx.ctx.Err()
|
|
}
|
|
if tx.isDone() {
|
|
return ErrTxDone
|
|
}
|
|
var err error
|
|
withLock(tx.dc, func() {
|
|
err = tx.txi.Commit()
|
|
})
|
|
if err != driver.ErrBadConn {
|
|
tx.closePrepared()
|
|
}
|
|
tx.close(err)
|
|
return err
|
|
}
|
|
|
|
// rollback aborts the transaction and optionally forces the pool to discard
|
|
// the connection.
|
|
func (tx *Tx) rollback(discardConn bool) error {
|
|
if tx.isDone() {
|
|
return ErrTxDone
|
|
}
|
|
var err error
|
|
withLock(tx.dc, func() {
|
|
err = tx.txi.Rollback()
|
|
})
|
|
if err != driver.ErrBadConn {
|
|
tx.closePrepared()
|
|
}
|
|
if discardConn {
|
|
err = driver.ErrBadConn
|
|
}
|
|
tx.close(err)
|
|
return err
|
|
}
|
|
|
|
// Rollback aborts the transaction.
|
|
func (tx *Tx) Rollback() error {
|
|
return tx.rollback(false)
|
|
}
|
|
|
|
// Prepare creates a prepared statement for use within a transaction.
|
|
//
|
|
// The returned statement operates within the transaction and will be closed
|
|
// when the transaction has been committed or rolled back.
|
|
//
|
|
// To use an existing prepared statement on this transaction, see Tx.Stmt.
|
|
//
|
|
// The provided context will be used for the preparation of the context, not
|
|
// for the execution of the returned statement. The returned statement
|
|
// will run in the transaction context.
|
|
func (tx *Tx) PrepareContext(ctx context.Context, query string) (*Stmt, error) {
|
|
// TODO(bradfitz): We could be more efficient here and either
|
|
// provide a method to take an existing Stmt (created on
|
|
// perhaps a different Conn), and re-create it on this Conn if
|
|
// necessary. Or, better: keep a map in DB of query string to
|
|
// Stmts, and have Stmt.Execute do the right thing and
|
|
// re-prepare if the Conn in use doesn't have that prepared
|
|
// statement. But we'll want to avoid caching the statement
|
|
// in the case where we only call conn.Prepare implicitly
|
|
// (such as in db.Exec or tx.Exec), but the caller package
|
|
// can't be holding a reference to the returned statement.
|
|
// Perhaps just looking at the reference count (by noting
|
|
// Stmt.Close) would be enough. We might also want a finalizer
|
|
// on Stmt to drop the reference count.
|
|
dc, err := tx.grabConn(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var si driver.Stmt
|
|
withLock(dc, func() {
|
|
si, err = ctxDriverPrepare(ctx, dc.ci, query)
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
stmt := &Stmt{
|
|
db: tx.db,
|
|
tx: tx,
|
|
txds: &driverStmt{
|
|
Locker: dc,
|
|
si: si,
|
|
},
|
|
query: query,
|
|
}
|
|
tx.stmts.Lock()
|
|
tx.stmts.v = append(tx.stmts.v, stmt)
|
|
tx.stmts.Unlock()
|
|
return stmt, nil
|
|
}
|
|
|
|
// Prepare creates a prepared statement for use within a transaction.
|
|
//
|
|
// The returned statement operates within the transaction and can no longer
|
|
// be used once the transaction has been committed or rolled back.
|
|
//
|
|
// To use an existing prepared statement on this transaction, see Tx.Stmt.
|
|
func (tx *Tx) Prepare(query string) (*Stmt, error) {
|
|
return tx.PrepareContext(context.Background(), query)
|
|
}
|
|
|
|
// StmtContext returns a transaction-specific prepared statement from
|
|
// an existing statement.
|
|
//
|
|
// Example:
|
|
// updateMoney, err := db.Prepare("UPDATE balance SET money=money+? WHERE id=?")
|
|
// ...
|
|
// tx, err := db.Begin()
|
|
// ...
|
|
// res, err := tx.StmtContext(ctx, updateMoney).Exec(123.45, 98293203)
|
|
//
|
|
// The returned statement operates within the transaction and will be closed
|
|
// when the transaction has been committed or rolled back.
|
|
func (tx *Tx) StmtContext(ctx context.Context, stmt *Stmt) *Stmt {
|
|
// TODO(bradfitz): optimize this. Currently this re-prepares
|
|
// each time. This is fine for now to illustrate the API but
|
|
// we should really cache already-prepared statements
|
|
// per-Conn. See also the big comment in Tx.Prepare.
|
|
|
|
if tx.db != stmt.db {
|
|
return &Stmt{stickyErr: errors.New("sql: Tx.Stmt: statement from different database used")}
|
|
}
|
|
dc, err := tx.grabConn(ctx)
|
|
if err != nil {
|
|
return &Stmt{stickyErr: err}
|
|
}
|
|
var si driver.Stmt
|
|
withLock(dc, func() {
|
|
si, err = ctxDriverPrepare(ctx, dc.ci, stmt.query)
|
|
})
|
|
txs := &Stmt{
|
|
db: tx.db,
|
|
tx: tx,
|
|
txds: &driverStmt{
|
|
Locker: dc,
|
|
si: si,
|
|
},
|
|
query: stmt.query,
|
|
stickyErr: err,
|
|
}
|
|
tx.stmts.Lock()
|
|
tx.stmts.v = append(tx.stmts.v, txs)
|
|
tx.stmts.Unlock()
|
|
return txs
|
|
}
|
|
|
|
// Stmt returns a transaction-specific prepared statement from
|
|
// an existing statement.
|
|
//
|
|
// Example:
|
|
// updateMoney, err := db.Prepare("UPDATE balance SET money=money+? WHERE id=?")
|
|
// ...
|
|
// tx, err := db.Begin()
|
|
// ...
|
|
// res, err := tx.Stmt(updateMoney).Exec(123.45, 98293203)
|
|
//
|
|
// The returned statement operates within the transaction and will be closed
|
|
// when the transaction has been committed or rolled back.
|
|
func (tx *Tx) Stmt(stmt *Stmt) *Stmt {
|
|
return tx.StmtContext(context.Background(), stmt)
|
|
}
|
|
|
|
// ExecContext executes a query that doesn't return rows.
|
|
// For example: an INSERT and UPDATE.
|
|
func (tx *Tx) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error) {
|
|
dc, err := tx.grabConn(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if execer, ok := dc.ci.(driver.Execer); ok {
|
|
dargs, err := driverArgs(nil, args)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var resi driver.Result
|
|
withLock(dc, func() {
|
|
resi, err = ctxDriverExec(ctx, execer, query, dargs)
|
|
})
|
|
if err == nil {
|
|
return driverResult{dc, resi}, nil
|
|
}
|
|
if err != driver.ErrSkip {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
var si driver.Stmt
|
|
withLock(dc, func() {
|
|
si, err = ctxDriverPrepare(ctx, dc.ci, query)
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ds := &driverStmt{Locker: dc, si: si}
|
|
defer ds.Close()
|
|
|
|
return resultFromStatement(ctx, ds, args...)
|
|
}
|
|
|
|
// Exec executes a query that doesn't return rows.
|
|
// For example: an INSERT and UPDATE.
|
|
func (tx *Tx) Exec(query string, args ...interface{}) (Result, error) {
|
|
return tx.ExecContext(context.Background(), query, args...)
|
|
}
|
|
|
|
// QueryContext executes a query that returns rows, typically a SELECT.
|
|
func (tx *Tx) QueryContext(ctx context.Context, query string, args ...interface{}) (*Rows, error) {
|
|
dc, err := tx.grabConn(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
releaseConn := func(error) {}
|
|
return tx.db.queryConn(ctx, dc, releaseConn, query, args)
|
|
}
|
|
|
|
// Query executes a query that returns rows, typically a SELECT.
|
|
func (tx *Tx) Query(query string, args ...interface{}) (*Rows, error) {
|
|
return tx.QueryContext(context.Background(), query, args...)
|
|
}
|
|
|
|
// QueryRowContext executes a query that is expected to return at most one row.
|
|
// QueryRowContext always returns a non-nil value. Errors are deferred until
|
|
// Row's Scan method is called.
|
|
func (tx *Tx) QueryRowContext(ctx context.Context, query string, args ...interface{}) *Row {
|
|
rows, err := tx.QueryContext(ctx, query, args...)
|
|
return &Row{rows: rows, err: err}
|
|
}
|
|
|
|
// QueryRow executes a query that is expected to return at most one row.
|
|
// QueryRow always returns a non-nil value. Errors are deferred until
|
|
// Row's Scan method is called.
|
|
func (tx *Tx) QueryRow(query string, args ...interface{}) *Row {
|
|
return tx.QueryRowContext(context.Background(), query, args...)
|
|
}
|
|
|
|
// connStmt is a prepared statement on a particular connection.
|
|
type connStmt struct {
|
|
dc *driverConn
|
|
ds *driverStmt
|
|
}
|
|
|
|
// Stmt is a prepared statement.
|
|
// A Stmt is safe for concurrent use by multiple goroutines.
|
|
type Stmt struct {
|
|
// Immutable:
|
|
db *DB // where we came from
|
|
query string // that created the Stmt
|
|
stickyErr error // if non-nil, this error is returned for all operations
|
|
|
|
closemu sync.RWMutex // held exclusively during close, for read otherwise.
|
|
|
|
// If in a transaction, else both nil:
|
|
tx *Tx
|
|
txds *driverStmt
|
|
|
|
mu sync.Mutex // protects the rest of the fields
|
|
closed bool
|
|
|
|
// css is a list of underlying driver statement interfaces
|
|
// that are valid on particular connections. This is only
|
|
// used if tx == nil and one is found that has idle
|
|
// connections. If tx != nil, txsi is always used.
|
|
css []connStmt
|
|
|
|
// lastNumClosed is copied from db.numClosed when Stmt is created
|
|
// without tx and closed connections in css are removed.
|
|
lastNumClosed uint64
|
|
}
|
|
|
|
// ExecContext executes a prepared statement with the given arguments and
|
|
// returns a Result summarizing the effect of the statement.
|
|
func (s *Stmt) ExecContext(ctx context.Context, args ...interface{}) (Result, error) {
|
|
s.closemu.RLock()
|
|
defer s.closemu.RUnlock()
|
|
|
|
var res Result
|
|
for i := 0; i < maxBadConnRetries; i++ {
|
|
_, releaseConn, ds, err := s.connStmt(ctx)
|
|
if err != nil {
|
|
if err == driver.ErrBadConn {
|
|
continue
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
res, err = resultFromStatement(ctx, ds, args...)
|
|
releaseConn(err)
|
|
if err != driver.ErrBadConn {
|
|
return res, err
|
|
}
|
|
}
|
|
return nil, driver.ErrBadConn
|
|
}
|
|
|
|
// Exec executes a prepared statement with the given arguments and
|
|
// returns a Result summarizing the effect of the statement.
|
|
func (s *Stmt) Exec(args ...interface{}) (Result, error) {
|
|
return s.ExecContext(context.Background(), args...)
|
|
}
|
|
|
|
func driverNumInput(ds *driverStmt) int {
|
|
ds.Lock()
|
|
defer ds.Unlock() // in case NumInput panics
|
|
return ds.si.NumInput()
|
|
}
|
|
|
|
func resultFromStatement(ctx context.Context, ds *driverStmt, args ...interface{}) (Result, error) {
|
|
want := driverNumInput(ds)
|
|
|
|
// -1 means the driver doesn't know how to count the number of
|
|
// placeholders, so we won't sanity check input here and instead let the
|
|
// driver deal with errors.
|
|
if want != -1 && len(args) != want {
|
|
return nil, fmt.Errorf("sql: expected %d arguments, got %d", want, len(args))
|
|
}
|
|
|
|
dargs, err := driverArgs(ds, args)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
ds.Lock()
|
|
defer ds.Unlock()
|
|
|
|
resi, err := ctxDriverStmtExec(ctx, ds.si, dargs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return driverResult{ds.Locker, resi}, nil
|
|
}
|
|
|
|
// removeClosedStmtLocked removes closed conns in s.css.
|
|
//
|
|
// To avoid lock contention on DB.mu, we do it only when
|
|
// s.db.numClosed - s.lastNum is large enough.
|
|
func (s *Stmt) removeClosedStmtLocked() {
|
|
t := len(s.css)/2 + 1
|
|
if t > 10 {
|
|
t = 10
|
|
}
|
|
dbClosed := atomic.LoadUint64(&s.db.numClosed)
|
|
if dbClosed-s.lastNumClosed < uint64(t) {
|
|
return
|
|
}
|
|
|
|
s.db.mu.Lock()
|
|
for i := 0; i < len(s.css); i++ {
|
|
if s.css[i].dc.dbmuClosed {
|
|
s.css[i] = s.css[len(s.css)-1]
|
|
s.css = s.css[:len(s.css)-1]
|
|
i--
|
|
}
|
|
}
|
|
s.db.mu.Unlock()
|
|
s.lastNumClosed = dbClosed
|
|
}
|
|
|
|
// connStmt returns a free driver connection on which to execute the
|
|
// statement, a function to call to release the connection, and a
|
|
// statement bound to that connection.
|
|
func (s *Stmt) connStmt(ctx context.Context) (ci *driverConn, releaseConn func(error), ds *driverStmt, err error) {
|
|
if err = s.stickyErr; err != nil {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
if s.closed {
|
|
s.mu.Unlock()
|
|
err = errors.New("sql: statement is closed")
|
|
return
|
|
}
|
|
|
|
// In a transaction, we always use the connection that the
|
|
// transaction was created on.
|
|
if s.tx != nil {
|
|
s.mu.Unlock()
|
|
ci, err = s.tx.grabConn(ctx) // blocks, waiting for the connection.
|
|
if err != nil {
|
|
return
|
|
}
|
|
releaseConn = func(error) {}
|
|
return ci, releaseConn, s.txds, nil
|
|
}
|
|
|
|
s.removeClosedStmtLocked()
|
|
s.mu.Unlock()
|
|
|
|
dc, err := s.db.conn(ctx, cachedOrNewConn)
|
|
if err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
|
|
s.mu.Lock()
|
|
for _, v := range s.css {
|
|
if v.dc == dc {
|
|
s.mu.Unlock()
|
|
return dc, dc.releaseConn, v.ds, nil
|
|
}
|
|
}
|
|
s.mu.Unlock()
|
|
|
|
// No luck; we need to prepare the statement on this connection
|
|
withLock(dc, func() {
|
|
ds, err = dc.prepareLocked(ctx, s.query)
|
|
})
|
|
if err != nil {
|
|
s.db.putConn(dc, err)
|
|
return nil, nil, nil, err
|
|
}
|
|
s.mu.Lock()
|
|
cs := connStmt{dc, ds}
|
|
s.css = append(s.css, cs)
|
|
s.mu.Unlock()
|
|
|
|
return dc, dc.releaseConn, ds, nil
|
|
}
|
|
|
|
// QueryContext executes a prepared query statement with the given arguments
|
|
// and returns the query results as a *Rows.
|
|
func (s *Stmt) QueryContext(ctx context.Context, args ...interface{}) (*Rows, error) {
|
|
s.closemu.RLock()
|
|
defer s.closemu.RUnlock()
|
|
|
|
var rowsi driver.Rows
|
|
for i := 0; i < maxBadConnRetries; i++ {
|
|
dc, releaseConn, ds, err := s.connStmt(ctx)
|
|
if err != nil {
|
|
if err == driver.ErrBadConn {
|
|
continue
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
rowsi, err = rowsiFromStatement(ctx, ds, args...)
|
|
if err == nil {
|
|
// Note: ownership of ci passes to the *Rows, to be freed
|
|
// with releaseConn.
|
|
rows := &Rows{
|
|
dc: dc,
|
|
rowsi: rowsi,
|
|
// releaseConn set below
|
|
}
|
|
rows.initContextClose(ctx)
|
|
s.db.addDep(s, rows)
|
|
rows.releaseConn = func(err error) {
|
|
releaseConn(err)
|
|
s.db.removeDep(s, rows)
|
|
}
|
|
return rows, nil
|
|
}
|
|
|
|
releaseConn(err)
|
|
if err != driver.ErrBadConn {
|
|
return nil, err
|
|
}
|
|
}
|
|
return nil, driver.ErrBadConn
|
|
}
|
|
|
|
// Query executes a prepared query statement with the given arguments
|
|
// and returns the query results as a *Rows.
|
|
func (s *Stmt) Query(args ...interface{}) (*Rows, error) {
|
|
return s.QueryContext(context.Background(), args...)
|
|
}
|
|
|
|
func rowsiFromStatement(ctx context.Context, ds *driverStmt, args ...interface{}) (driver.Rows, error) {
|
|
var want int
|
|
withLock(ds, func() {
|
|
want = ds.si.NumInput()
|
|
})
|
|
|
|
// -1 means the driver doesn't know how to count the number of
|
|
// placeholders, so we won't sanity check input here and instead let the
|
|
// driver deal with errors.
|
|
if want != -1 && len(args) != want {
|
|
return nil, fmt.Errorf("sql: statement expects %d inputs; got %d", want, len(args))
|
|
}
|
|
|
|
dargs, err := driverArgs(ds, args)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
ds.Lock()
|
|
defer ds.Unlock()
|
|
|
|
rowsi, err := ctxDriverStmtQuery(ctx, ds.si, dargs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return rowsi, nil
|
|
}
|
|
|
|
// QueryRowContext executes a prepared query statement with the given arguments.
|
|
// If an error occurs during the execution of the statement, that error will
|
|
// be returned by a call to Scan on the returned *Row, which is always non-nil.
|
|
// If the query selects no rows, the *Row's Scan will return ErrNoRows.
|
|
// Otherwise, the *Row's Scan scans the first selected row and discards
|
|
// the rest.
|
|
//
|
|
// Example usage:
|
|
//
|
|
// var name string
|
|
// err := nameByUseridStmt.QueryRowContext(ctx, id).Scan(&name)
|
|
func (s *Stmt) QueryRowContext(ctx context.Context, args ...interface{}) *Row {
|
|
rows, err := s.QueryContext(ctx, args...)
|
|
if err != nil {
|
|
return &Row{err: err}
|
|
}
|
|
return &Row{rows: rows}
|
|
}
|
|
|
|
// QueryRow executes a prepared query statement with the given arguments.
|
|
// If an error occurs during the execution of the statement, that error will
|
|
// be returned by a call to Scan on the returned *Row, which is always non-nil.
|
|
// If the query selects no rows, the *Row's Scan will return ErrNoRows.
|
|
// Otherwise, the *Row's Scan scans the first selected row and discards
|
|
// the rest.
|
|
//
|
|
// Example usage:
|
|
//
|
|
// var name string
|
|
// err := nameByUseridStmt.QueryRow(id).Scan(&name)
|
|
func (s *Stmt) QueryRow(args ...interface{}) *Row {
|
|
return s.QueryRowContext(context.Background(), args...)
|
|
}
|
|
|
|
// Close closes the statement.
|
|
func (s *Stmt) Close() error {
|
|
s.closemu.Lock()
|
|
defer s.closemu.Unlock()
|
|
|
|
if s.stickyErr != nil {
|
|
return s.stickyErr
|
|
}
|
|
s.mu.Lock()
|
|
if s.closed {
|
|
s.mu.Unlock()
|
|
return nil
|
|
}
|
|
s.closed = true
|
|
s.mu.Unlock()
|
|
|
|
if s.tx != nil {
|
|
return s.txds.Close()
|
|
}
|
|
|
|
return s.db.removeDep(s, s)
|
|
}
|
|
|
|
func (s *Stmt) finalClose() error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.css != nil {
|
|
for _, v := range s.css {
|
|
s.db.noteUnusedDriverStatement(v.dc, v.ds)
|
|
v.dc.removeOpenStmt(v.ds)
|
|
}
|
|
s.css = nil
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Rows is the result of a query. Its cursor starts before the first row
|
|
// of the result set. Use Next to advance through the rows:
|
|
//
|
|
// rows, err := db.Query("SELECT ...")
|
|
// ...
|
|
// defer rows.Close()
|
|
// for rows.Next() {
|
|
// var id int
|
|
// var name string
|
|
// err = rows.Scan(&id, &name)
|
|
// ...
|
|
// }
|
|
// err = rows.Err() // get any error encountered during iteration
|
|
// ...
|
|
type Rows struct {
|
|
dc *driverConn // owned; must call releaseConn when closed to release
|
|
releaseConn func(error)
|
|
rowsi driver.Rows
|
|
|
|
// closed value is 1 when the Rows is closed.
|
|
// Use atomic operations on value when checking value.
|
|
closed int32
|
|
ctxClose chan struct{} // closed when Rows is closed, may be null.
|
|
lastcols []driver.Value
|
|
lasterr error // non-nil only if closed is true
|
|
closeStmt *driverStmt // if non-nil, statement to Close on close
|
|
}
|
|
|
|
func (rs *Rows) initContextClose(ctx context.Context) {
|
|
if ctx.Done() == context.Background().Done() {
|
|
return
|
|
}
|
|
|
|
rs.ctxClose = make(chan struct{})
|
|
go func() {
|
|
select {
|
|
case <-ctx.Done():
|
|
rs.Close()
|
|
case <-rs.ctxClose:
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Next prepares the next result row for reading with the Scan method. It
|
|
// returns true on success, or false if there is no next result row or an error
|
|
// happened while preparing it. Err should be consulted to distinguish between
|
|
// the two cases.
|
|
//
|
|
// Every call to Scan, even the first one, must be preceded by a call to Next.
|
|
func (rs *Rows) Next() bool {
|
|
if rs.isClosed() {
|
|
return false
|
|
}
|
|
if rs.lastcols == nil {
|
|
rs.lastcols = make([]driver.Value, len(rs.rowsi.Columns()))
|
|
}
|
|
rs.lasterr = rs.rowsi.Next(rs.lastcols)
|
|
if rs.lasterr != nil {
|
|
// Close the connection if there is a driver error.
|
|
if rs.lasterr != io.EOF {
|
|
rs.Close()
|
|
return false
|
|
}
|
|
nextResultSet, ok := rs.rowsi.(driver.RowsNextResultSet)
|
|
if !ok {
|
|
rs.Close()
|
|
return false
|
|
}
|
|
// The driver is at the end of the current result set.
|
|
// Test to see if there is another result set after the current one.
|
|
// Only close Rows if there is no futher result sets to read.
|
|
if !nextResultSet.HasNextResultSet() {
|
|
rs.Close()
|
|
}
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// NextResultSet prepares the next result set for reading. It returns true if
|
|
// there is further result sets, or false if there is no further result set
|
|
// or if there is an error advancing to it. The Err method should be consulted
|
|
// to distinguish between the two cases.
|
|
//
|
|
// After calling NextResultSet, the Next method should always be called before
|
|
// scanning. If there are further result sets they may not have rows in the result
|
|
// set.
|
|
func (rs *Rows) NextResultSet() bool {
|
|
if rs.isClosed() {
|
|
return false
|
|
}
|
|
rs.lastcols = nil
|
|
nextResultSet, ok := rs.rowsi.(driver.RowsNextResultSet)
|
|
if !ok {
|
|
rs.Close()
|
|
return false
|
|
}
|
|
rs.lasterr = nextResultSet.NextResultSet()
|
|
if rs.lasterr != nil {
|
|
rs.Close()
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// Err returns the error, if any, that was encountered during iteration.
|
|
// Err may be called after an explicit or implicit Close.
|
|
func (rs *Rows) Err() error {
|
|
if rs.lasterr == io.EOF {
|
|
return nil
|
|
}
|
|
return rs.lasterr
|
|
}
|
|
|
|
// Columns returns the column names.
|
|
// Columns returns an error if the rows are closed, or if the rows
|
|
// are from QueryRow and there was a deferred error.
|
|
func (rs *Rows) Columns() ([]string, error) {
|
|
if rs.isClosed() {
|
|
return nil, errors.New("sql: Rows are closed")
|
|
}
|
|
if rs.rowsi == nil {
|
|
return nil, errors.New("sql: no Rows available")
|
|
}
|
|
return rs.rowsi.Columns(), nil
|
|
}
|
|
|
|
// ColumnTypes returns column information such as column type, length,
|
|
// and nullable. Some information may not be available from some drivers.
|
|
func (rs *Rows) ColumnTypes() ([]*ColumnType, error) {
|
|
if rs.isClosed() {
|
|
return nil, errors.New("sql: Rows are closed")
|
|
}
|
|
if rs.rowsi == nil {
|
|
return nil, errors.New("sql: no Rows available")
|
|
}
|
|
return rowsColumnInfoSetup(rs.rowsi), nil
|
|
}
|
|
|
|
// ColumnType contains the name and type of a column.
|
|
type ColumnType struct {
|
|
name string
|
|
|
|
hasNullable bool
|
|
hasLength bool
|
|
hasPrecisionScale bool
|
|
|
|
nullable bool
|
|
length int64
|
|
databaseType string
|
|
precision int64
|
|
scale int64
|
|
scanType reflect.Type
|
|
}
|
|
|
|
// Name returns the name or alias of the column.
|
|
func (ci *ColumnType) Name() string {
|
|
return ci.name
|
|
}
|
|
|
|
// Length returns the column type length for variable length column types such
|
|
// as text and binary field types. If the type length is unbounded the value will
|
|
// be math.MaxInt64 (any database limits will still apply).
|
|
// If the column type is not variable length, such as an int, or if not supported
|
|
// by the driver ok is false.
|
|
func (ci *ColumnType) Length() (length int64, ok bool) {
|
|
return ci.length, ci.hasLength
|
|
}
|
|
|
|
// DecimalSize returns the scale and precision of a decimal type.
|
|
// If not applicable or if not supported ok is false.
|
|
func (ci *ColumnType) DecimalSize() (precision, scale int64, ok bool) {
|
|
return ci.precision, ci.scale, ci.hasPrecisionScale
|
|
}
|
|
|
|
// ScanType returns a Go type suitable for scanning into using Rows.Scan.
|
|
// If a driver does not support this property ScanType will return
|
|
// the type of an empty interface.
|
|
func (ci *ColumnType) ScanType() reflect.Type {
|
|
return ci.scanType
|
|
}
|
|
|
|
// Nullable returns whether the column may be null.
|
|
// If a driver does not support this property ok will be false.
|
|
func (ci *ColumnType) Nullable() (nullable, ok bool) {
|
|
return ci.nullable, ci.hasNullable
|
|
}
|
|
|
|
// DatabaseTypeName returns the database system name of the column type. If an empty
|
|
// string is returned the driver type name is not supported.
|
|
// Consult your driver documentation for a list of driver data types. Length specifiers
|
|
// are not included.
|
|
// Common type include "VARCHAR", "TEXT", "NVARCHAR", "DECIMAL", "BOOL", "INT", "BIGINT".
|
|
func (ci *ColumnType) DatabaseTypeName() string {
|
|
return ci.databaseType
|
|
}
|
|
|
|
func rowsColumnInfoSetup(rowsi driver.Rows) []*ColumnType {
|
|
names := rowsi.Columns()
|
|
|
|
list := make([]*ColumnType, len(names))
|
|
for i := range list {
|
|
ci := &ColumnType{
|
|
name: names[i],
|
|
}
|
|
list[i] = ci
|
|
|
|
if prop, ok := rowsi.(driver.RowsColumnTypeScanType); ok {
|
|
ci.scanType = prop.ColumnTypeScanType(i)
|
|
} else {
|
|
ci.scanType = reflect.TypeOf(new(interface{})).Elem()
|
|
}
|
|
if prop, ok := rowsi.(driver.RowsColumnTypeDatabaseTypeName); ok {
|
|
ci.databaseType = prop.ColumnTypeDatabaseTypeName(i)
|
|
}
|
|
if prop, ok := rowsi.(driver.RowsColumnTypeLength); ok {
|
|
ci.length, ci.hasLength = prop.ColumnTypeLength(i)
|
|
}
|
|
if prop, ok := rowsi.(driver.RowsColumnTypeNullable); ok {
|
|
ci.nullable, ci.hasNullable = prop.ColumnTypeNullable(i)
|
|
}
|
|
if prop, ok := rowsi.(driver.RowsColumnTypePrecisionScale); ok {
|
|
ci.precision, ci.scale, ci.hasPrecisionScale = prop.ColumnTypePrecisionScale(i)
|
|
}
|
|
}
|
|
return list
|
|
}
|
|
|
|
// Scan copies the columns in the current row into the values pointed
|
|
// at by dest. The number of values in dest must be the same as the
|
|
// number of columns in Rows.
|
|
//
|
|
// Scan converts columns read from the database into the following
|
|
// common Go types and special types provided by the sql package:
|
|
//
|
|
// *string
|
|
// *[]byte
|
|
// *int, *int8, *int16, *int32, *int64
|
|
// *uint, *uint8, *uint16, *uint32, *uint64
|
|
// *bool
|
|
// *float32, *float64
|
|
// *interface{}
|
|
// *RawBytes
|
|
// any type implementing Scanner (see Scanner docs)
|
|
//
|
|
// In the most simple case, if the type of the value from the source
|
|
// column is an integer, bool or string type T and dest is of type *T,
|
|
// Scan simply assigns the value through the pointer.
|
|
//
|
|
// Scan also converts between string and numeric types, as long as no
|
|
// information would be lost. While Scan stringifies all numbers
|
|
// scanned from numeric database columns into *string, scans into
|
|
// numeric types are checked for overflow. For example, a float64 with
|
|
// value 300 or a string with value "300" can scan into a uint16, but
|
|
// not into a uint8, though float64(255) or "255" can scan into a
|
|
// uint8. One exception is that scans of some float64 numbers to
|
|
// strings may lose information when stringifying. In general, scan
|
|
// floating point columns into *float64.
|
|
//
|
|
// If a dest argument has type *[]byte, Scan saves in that argument a
|
|
// copy of the corresponding data. The copy is owned by the caller and
|
|
// can be modified and held indefinitely. The copy can be avoided by
|
|
// using an argument of type *RawBytes instead; see the documentation
|
|
// for RawBytes for restrictions on its use.
|
|
//
|
|
// If an argument has type *interface{}, Scan copies the value
|
|
// provided by the underlying driver without conversion. When scanning
|
|
// from a source value of type []byte to *interface{}, a copy of the
|
|
// slice is made and the caller owns the result.
|
|
//
|
|
// Source values of type time.Time may be scanned into values of type
|
|
// *time.Time, *interface{}, *string, or *[]byte. When converting to
|
|
// the latter two, time.Format3339Nano is used.
|
|
//
|
|
// Source values of type bool may be scanned into types *bool,
|
|
// *interface{}, *string, *[]byte, or *RawBytes.
|
|
//
|
|
// For scanning into *bool, the source may be true, false, 1, 0, or
|
|
// string inputs parseable by strconv.ParseBool.
|
|
func (rs *Rows) Scan(dest ...interface{}) error {
|
|
if rs.isClosed() {
|
|
return errors.New("sql: Rows are closed")
|
|
}
|
|
if rs.lastcols == nil {
|
|
return errors.New("sql: Scan called without calling Next")
|
|
}
|
|
if len(dest) != len(rs.lastcols) {
|
|
return fmt.Errorf("sql: expected %d destination arguments in Scan, not %d", len(rs.lastcols), len(dest))
|
|
}
|
|
for i, sv := range rs.lastcols {
|
|
err := convertAssign(dest[i], sv)
|
|
if err != nil {
|
|
return fmt.Errorf("sql: Scan error on column index %d: %v", i, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var rowsCloseHook func(*Rows, *error)
|
|
|
|
func (rs *Rows) isClosed() bool {
|
|
return atomic.LoadInt32(&rs.closed) != 0
|
|
}
|
|
|
|
// Close closes the Rows, preventing further enumeration. If Next is called
|
|
// and returns false and there are no further result sets,
|
|
// the Rows are closed automatically and it will suffice to check the
|
|
// result of Err. Close is idempotent and does not affect the result of Err.
|
|
func (rs *Rows) Close() error {
|
|
if !atomic.CompareAndSwapInt32(&rs.closed, 0, 1) {
|
|
return nil
|
|
}
|
|
if rs.ctxClose != nil {
|
|
close(rs.ctxClose)
|
|
}
|
|
err := rs.rowsi.Close()
|
|
if fn := rowsCloseHook; fn != nil {
|
|
fn(rs, &err)
|
|
}
|
|
if rs.closeStmt != nil {
|
|
rs.closeStmt.Close()
|
|
}
|
|
rs.releaseConn(err)
|
|
return err
|
|
}
|
|
|
|
// Row is the result of calling QueryRow to select a single row.
|
|
type Row struct {
|
|
// One of these two will be non-nil:
|
|
err error // deferred error for easy chaining
|
|
rows *Rows
|
|
}
|
|
|
|
// Scan copies the columns from the matched row into the values
|
|
// pointed at by dest. See the documentation on Rows.Scan for details.
|
|
// If more than one row matches the query,
|
|
// Scan uses the first row and discards the rest. If no row matches
|
|
// the query, Scan returns ErrNoRows.
|
|
func (r *Row) Scan(dest ...interface{}) error {
|
|
if r.err != nil {
|
|
return r.err
|
|
}
|
|
|
|
// TODO(bradfitz): for now we need to defensively clone all
|
|
// []byte that the driver returned (not permitting
|
|
// *RawBytes in Rows.Scan), since we're about to close
|
|
// the Rows in our defer, when we return from this function.
|
|
// the contract with the driver.Next(...) interface is that it
|
|
// can return slices into read-only temporary memory that's
|
|
// only valid until the next Scan/Close. But the TODO is that
|
|
// for a lot of drivers, this copy will be unnecessary. We
|
|
// should provide an optional interface for drivers to
|
|
// implement to say, "don't worry, the []bytes that I return
|
|
// from Next will not be modified again." (for instance, if
|
|
// they were obtained from the network anyway) But for now we
|
|
// don't care.
|
|
defer r.rows.Close()
|
|
for _, dp := range dest {
|
|
if _, ok := dp.(*RawBytes); ok {
|
|
return errors.New("sql: RawBytes isn't allowed on Row.Scan")
|
|
}
|
|
}
|
|
|
|
if !r.rows.Next() {
|
|
if err := r.rows.Err(); err != nil {
|
|
return err
|
|
}
|
|
return ErrNoRows
|
|
}
|
|
err := r.rows.Scan(dest...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Make sure the query can be processed to completion with no errors.
|
|
if err := r.rows.Close(); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// A Result summarizes an executed SQL command.
|
|
type Result interface {
|
|
// LastInsertId returns the integer generated by the database
|
|
// in response to a command. Typically this will be from an
|
|
// "auto increment" column when inserting a new row. Not all
|
|
// databases support this feature, and the syntax of such
|
|
// statements varies.
|
|
LastInsertId() (int64, error)
|
|
|
|
// RowsAffected returns the number of rows affected by an
|
|
// update, insert, or delete. Not every database or database
|
|
// driver may support this.
|
|
RowsAffected() (int64, error)
|
|
}
|
|
|
|
type driverResult struct {
|
|
sync.Locker // the *driverConn
|
|
resi driver.Result
|
|
}
|
|
|
|
func (dr driverResult) LastInsertId() (int64, error) {
|
|
dr.Lock()
|
|
defer dr.Unlock()
|
|
return dr.resi.LastInsertId()
|
|
}
|
|
|
|
func (dr driverResult) RowsAffected() (int64, error) {
|
|
dr.Lock()
|
|
defer dr.Unlock()
|
|
return dr.resi.RowsAffected()
|
|
}
|
|
|
|
func stack() string {
|
|
var buf [2 << 10]byte
|
|
return string(buf[:runtime.Stack(buf[:], false)])
|
|
}
|
|
|
|
// withLock runs while holding lk.
|
|
func withLock(lk sync.Locker, fn func()) {
|
|
lk.Lock()
|
|
defer lk.Unlock() // in case fn panics
|
|
fn()
|
|
}
|