Centralize access to connection pools

Author: dustin

client.go

@@ -48,10 +48,12 @@ var MaxBulkRetries = 10
 // your command will only be executed only once.
 func (b *Bucket) Do(k string, f func(mc *memcached.Client, vb uint16) error) error {
 	vb := b.VBHash(k)
-	for i := 0; i < len(b.connections)*2; i++ {
+	maxTries := len(b.VBucketServerMap.ServerList) * 2
+	for i := 0; i < maxTries; i++ {
 		masterId := b.VBucketServerMap.VBucketMap[vb][0]
-		conn, err := b.connections[masterId].Get()
-		defer b.connections[masterId].Return(conn)
+		pool := b.getConnPool(masterId)
+		conn, err := pool.Get()
+		defer pool.Return(conn)
 		if err != nil {
 			return err
 		}
@@ -71,8 +73,8 @@ func (b *Bucket) Do(k string, f func(mc *memcached.Client, vb uint16) error) err
 		}
 	}
 
-	return fmt.Errorf("Unable to complete action after %v retries",
-		len(b.connections)*2)
+	return fmt.Errorf("Unable to complete action after %v attemps",
+		maxTries)
 }
 
 type gathered_stats struct {
@@ -85,8 +87,9 @@ func getStatsParallel(b *Bucket, offset int, which string,
 	sn := b.VBucketServerMap.ServerList[offset]
 
 	results := map[string]string{}
-	conn, err := b.connections[offset].Get()
-	defer b.connections[offset].Return(conn)
+	pool := b.getConnPool(offset)
+	conn, err := pool.Get()
+	defer pool.Return(conn)
 	if err != nil {
 		ch <- gathered_stats{sn, results}
 	} else {
@@ -151,12 +154,13 @@ func (b *Bucket) doBulkGet(vb uint16, keys []string,
 		// This stack frame exists to ensure we can clean up
 		// connection at a reasonable time.
 		err := func() error {
-			conn, err := b.connections[masterId].Get()
+			pool := b.getConnPool(masterId)
+			conn, err := pool.Get()
 			if err != nil {
 				ch <- map[string]*gomemcached.MCResponse{}
 				return err
 			}
-			defer b.connections[masterId].Return(conn)
+			defer pool.Return(conn)
 
 			m, err := conn.GetBulk(vb, keys)
 			switch err.(type) {

pools.go

@@ -102,10 +102,22 @@ type Bucket struct {
 	Controllers map[string]interface{} `json:"controllers,omitempty"`
 
 	pool        *Pool
-	connections []*connectionPool
+	connPools   []*connectionPool
 	commonSufix string
 }
 
+func (b Bucket) getConnPools() []*connectionPool {
+	return b.connPools
+}
+
+func (b Bucket) getConnPool(i int) *connectionPool {
+	p := b.getConnPools()
+	if len(p) > i {
+		return p[i]
+	}
+	return nil
+}
+
 func (b Bucket) authHandler() (ah AuthHandler) {
 	if b.pool != nil {
 		ah = b.pool.client.ah
@@ -228,8 +240,8 @@ func (b *Bucket) refresh() (err error) {
 		return err
 	}
 	b.pool = pool
-	for i := range b.connections {
-		b.connections[i] = newConnectionPool(
+	for i := range b.connPools {
+		b.connPools[i] = newConnectionPool(
 			b.VBucketServerMap.ServerList[i],
 			b.authHandler(), PoolSize, PoolOverflow)
 	}
@@ -246,7 +258,7 @@ func (p *Pool) refresh() (err error) {
 	}
 	for _, b := range buckets {
 		b.pool = p
-		b.connections = make([]*connectionPool, len(b.VBucketServerMap.ServerList))
+		b.connPools = make([]*connectionPool, len(b.VBucketServerMap.ServerList))
 
 		p.BucketMap[b.Name] = b
 	}
@@ -275,18 +287,18 @@ func (c *Client) GetPool(name string) (p Pool, err error) {
 
 // Mark this bucket as no longer needed, closing connections it may have open.
 func (b *Bucket) Close() {
-	if b.connections != nil {
-		for _, c := range b.connections {
+	if b.connPools != nil {
+		for _, c := range b.connPools {
 			if c != nil {
 				c.Close()
 			}
 		}
-		b.connections = nil
+		b.connPools = nil
 	}
 }
 
 func bucket_finalizer(b *Bucket) {
-	if b.connections != nil {
+	if b.connPools != nil {
 		log.Printf("Warning: Finalizing a bucket with active connections.")
 	}
 }

tap.go

@@ -81,7 +81,7 @@ func (feed *TapFeed) run() {
 
 func (feed *TapFeed) connectToNodes() (killSwitch chan bool, err error) {
 	killSwitch = make(chan bool)
-	for _, serverConn := range feed.bucket.connections {
+	for _, serverConn := range feed.bucket.getConnPools() {
 		var singleFeed *memcached.TapFeed
 		singleFeed, err = serverConn.StartTapFeed(feed.args)
 		if err != nil {