| // 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" |
| "errors" |
| "fmt" |
| "io" |
| "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 |
| } |
| |
| // 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[driver.Stmt]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(si driver.Stmt) { |
| dc.Lock() |
| defer dc.Unlock() |
| delete(dc.openStmt, si) |
| } |
| |
| 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) (driver.Stmt, error) { |
| si, err := ctxDriverPrepare(ctx, dc.ci, query) |
| if err == nil { |
| // Track each driverConn's open statements, so we can close them |
| // before closing the conn. |
| // |
| // TODO(bradfitz): let drivers opt out of caring about |
| // stmt closes if the conn is about to close anyway? For now |
| // do the safe thing, in case stmts need to be closed. |
| // |
| // TODO(bradfitz): after Go 1.2, closing driver.Stmts |
| // should be moved to driverStmt, using unique |
| // *driverStmts everywhere (including from |
| // *Stmt.connStmt, instead of returning a |
| // driver.Stmt), using driverStmt as a pointer |
| // everywhere, and making it a finalCloser. |
| if dc.openStmt == nil { |
| dc.openStmt = make(map[driver.Stmt]bool) |
| } |
| dc.openStmt[si] = true |
| } |
| return si, err |
| } |
| |
| // 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 { |
| dc.Lock() |
| |
| for si := range dc.openStmt { |
| si.Close() |
| } |
| dc.openStmt = nil |
| |
| err := dc.ci.Close() |
| dc.ci = nil |
| dc.finalClosed = true |
| dc.Unlock() |
| |
| 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 |
| } |
| |
| func (ds *driverStmt) Close() error { |
| ds.Lock() |
| defer ds.Unlock() |
| return ds.si.Close() |
| } |
| |
| // 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 { |
| // TODO(bradfitz): give drivers an optional hook to implement |
| // this in a more efficient or more reliable way, if they |
| // have one. |
| dc, err := db.conn(ctx, cachedOrNewConn) |
| if err != nil { |
| return err |
| } |
| db.putConn(dc, nil) |
| return nil |
| } |
| |
| // 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. |
| if err := ctx.Err(); err != nil { |
| return nil, 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 si 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, si driver.Stmt) { |
| db.mu.Lock() |
| defer db.mu.Unlock() |
| if c.inUse { |
| c.onPut = append(c.onPut, func() { |
| si.Close() |
| }) |
| } else { |
| c.Lock() |
| defer c.Unlock() |
| if !c.finalClosed { |
| si.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 for the preparation of the statment, 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 si driver.Stmt |
| withLock(dc, func() { |
| si, err = dc.prepareLocked(ctx, query) |
| }) |
| if err != nil { |
| db.putConn(dc, err) |
| return nil, err |
| } |
| stmt := &Stmt{ |
| db: db, |
| query: query, |
| css: []connStmt{{dc, si}}, |
| 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.Value |
| 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 |
| } |
| defer withLock(dc, func() { si.Close() }) |
| return resultFromStatement(ctx, driverStmt{dc, si}, 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{dc, si} |
| rowsi, err := rowsiFromStatement(ctx, ds, args...) |
| if err != nil { |
| withLock(dc, func() { |
| si.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: si, |
| } |
| 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. 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, |
| } |
| go func() { |
| select { |
| case <-ctx.Done(): |
| if !tx.done { |
| tx.Rollback() |
| } |
| } |
| }() |
| 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 false to true exactly once, on Commit |
| // or Rollback. once done, all operations fail with |
| // ErrTxDone. |
| done bool |
| |
| // 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() |
| } |
| |
| // 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 tx.done { |
| panic("double close") // internal error |
| } |
| tx.done = true |
| 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.done { |
| 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 { |
| if tx.done { |
| 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. |
| func (tx *Tx) Rollback() error { |
| if tx.done { |
| return ErrTxDone |
| } |
| var err error |
| withLock(tx.dc, func() { |
| err = tx.txi.Rollback() |
| }) |
| if err != driver.ErrBadConn { |
| tx.closePrepared() |
| } |
| tx.close(err) |
| return err |
| } |
| |
| // 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. |
| // |
| // 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 = dc.ci.Prepare(query) |
| }) |
| if err != nil { |
| return nil, err |
| } |
| |
| stmt := &Stmt{ |
| db: tx.db, |
| tx: tx, |
| txsi: &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 can no longer |
| // be used once 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 = dc.ci.Prepare(stmt.query) |
| }) |
| txs := &Stmt{ |
| db: tx.db, |
| tx: tx, |
| txsi: &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 can no longer |
| // be used once 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 |
| } |
| defer withLock(dc, func() { si.Close() }) |
| |
| return resultFromStatement(ctx, driverStmt{dc, si}, 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 |
| si driver.Stmt |
| } |
| |
| // 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 |
| txsi *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++ { |
| dc, releaseConn, si, err := s.connStmt(ctx) |
| if err != nil { |
| if err == driver.ErrBadConn { |
| continue |
| } |
| return nil, err |
| } |
| |
| res, err = resultFromStatement(ctx, driverStmt{dc, si}, 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), si driver.Stmt, 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.txsi.si, 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.si, nil |
| } |
| } |
| s.mu.Unlock() |
| |
| // No luck; we need to prepare the statement on this connection |
| withLock(dc, func() { |
| si, 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, si} |
| s.css = append(s.css, cs) |
| s.mu.Unlock() |
| |
| return dc, dc.releaseConn, si, 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, si, err := s.connStmt(ctx) |
| if err != nil { |
| if err == driver.ErrBadConn { |
| continue |
| } |
| return nil, err |
| } |
| |
| rowsi, err = rowsiFromStatement(ctx, driverStmt{dc, si}, 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 |
| |
| if s.tx != nil { |
| defer s.mu.Unlock() |
| return s.txsi.Close() |
| } |
| s.mu.Unlock() |
| |
| 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.si) |
| v.dc.removeOpenStmt(v.si) |
| } |
| 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 driver.Stmt // 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 { |
| 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 |
| } |
| |
| // 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 returns |
| // false, 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() |
| } |