mirror of https://github.com/jackc/pgx.git
ConnPool no longer creates connections just to release them.
parent
efea92e8be
commit
d2ce556450
25
conn_pool.go
25
conn_pool.go
|
@ -19,6 +19,7 @@ type ConnPool struct {
|
||||||
maxConnections int
|
maxConnections int
|
||||||
afterConnect func(*Conn) error
|
afterConnect func(*Conn) error
|
||||||
logger Logger
|
logger Logger
|
||||||
|
closed bool
|
||||||
}
|
}
|
||||||
|
|
||||||
type ConnPoolStat struct {
|
type ConnPoolStat struct {
|
||||||
|
@ -68,6 +69,10 @@ func (p *ConnPool) Acquire() (c *Conn, err error) {
|
||||||
p.cond.L.Lock()
|
p.cond.L.Lock()
|
||||||
defer p.cond.L.Unlock()
|
defer p.cond.L.Unlock()
|
||||||
|
|
||||||
|
if p.closed {
|
||||||
|
return nil, errors.New("cannot acquire from closed pool")
|
||||||
|
}
|
||||||
|
|
||||||
// A connection is available
|
// A connection is available
|
||||||
if len(p.availableConnections) > 0 {
|
if len(p.availableConnections) > 0 {
|
||||||
c = p.availableConnections[len(p.availableConnections)-1]
|
c = p.availableConnections[len(p.availableConnections)-1]
|
||||||
|
@ -122,13 +127,25 @@ func (p *ConnPool) Release(conn *Conn) {
|
||||||
p.cond.Signal()
|
p.cond.Signal()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Close ends the use of a connection pool by closing all underlying connections.
|
// Close ends the use of a connection pool. It prevents any new connections
|
||||||
|
// from being acquired, waits until all acquired connections are released,
|
||||||
|
// then closes all underlying connections.
|
||||||
func (p *ConnPool) Close() {
|
func (p *ConnPool) Close() {
|
||||||
for i := 0; i < p.maxConnections; i++ {
|
p.cond.L.Lock()
|
||||||
if c, err := p.Acquire(); err == nil {
|
defer p.cond.L.Unlock()
|
||||||
_ = c.Close()
|
|
||||||
|
p.closed = true
|
||||||
|
|
||||||
|
// Wait until all connections are released
|
||||||
|
if len(p.availableConnections) != len(p.allConnections) {
|
||||||
|
for len(p.availableConnections) != len(p.allConnections) {
|
||||||
|
p.cond.Wait()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for _, c := range p.allConnections {
|
||||||
|
_ = c.Close()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stat returns connection pool statistics
|
// Stat returns connection pool statistics
|
||||||
|
|
|
@ -416,8 +416,12 @@ func TestConnPoolQueryConcurrentLoad(t *testing.T) {
|
||||||
pool := createConnPool(t, 10)
|
pool := createConnPool(t, 10)
|
||||||
defer pool.Close()
|
defer pool.Close()
|
||||||
|
|
||||||
for i := 0; i < 100; i++ {
|
n := 100
|
||||||
|
done := make(chan bool)
|
||||||
|
|
||||||
|
for i := 0; i < n; i++ {
|
||||||
go func() {
|
go func() {
|
||||||
|
defer func() { done <- true }()
|
||||||
var rowCount int32
|
var rowCount int32
|
||||||
|
|
||||||
rows, err := pool.Query("select generate_series(1,$1)", 1000)
|
rows, err := pool.Query("select generate_series(1,$1)", 1000)
|
||||||
|
@ -447,6 +451,10 @@ func TestConnPoolQueryConcurrentLoad(t *testing.T) {
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for i := 0; i < n; i++ {
|
||||||
|
<-done
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestConnPoolQueryRow(t *testing.T) {
|
func TestConnPoolQueryRow(t *testing.T) {
|
||||||
|
|
Loading…
Reference in New Issue