Permalink
Browse files

Atomically perform bucket refreshes

This should make things considerably less racy under failure.  Needs
more actual race testing with the race detector, though.

There are numerous caveats here since the bucket is still used both for
generating and parsing bucket descriptions on the wire *and* used to
represent a bucket within a running system.

The parts we expect to change may now do so freely.

cbugg: close bug-909 bug-903 bug-853
  • Loading branch information...
1 parent 794e447 commit 9a55f93f2827b9cd5583e0840d062fa44ebcd4f5 @dustin dustin committed Oct 24, 2013
Showing with 102 additions and 57 deletions.
  1. +10 −8 client.go
  2. +4 −4 examples/hello/hello.go
  3. +59 −21 pools.go
  4. +8 −6 pools_test.go
  5. +2 −1 vbmap.go
  6. +4 −2 vbmap_test.go
  7. +4 −3 views.go
  8. +11 −12 views_test.go
View
@@ -48,9 +48,10 @@ 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)
- maxTries := len(b.VBucketServerMap.ServerList) * 2
+ maxTries := len(b.Nodes()) * 2
for i := 0; i < maxTries; i++ {
- masterId := b.VBucketServerMap.VBucketMap[vb][0]
+ vbm := b.VBServerMap()
+ masterId := vbm.VBucketMap[vb][0]
pool := b.getConnPool(masterId)
conn, err := pool.Get()
defer pool.Return(conn)
@@ -84,7 +85,7 @@ type gathered_stats struct {
func getStatsParallel(b *Bucket, offset int, which string,
ch chan<- gathered_stats) {
- sn := b.VBucketServerMap.ServerList[offset]
+ sn := b.VBServerMap().ServerList[offset]
results := map[string]string{}
pool := b.getConnPool(offset)
@@ -108,19 +109,20 @@ func getStatsParallel(b *Bucket, offset int, which string,
func (b *Bucket) GetStats(which string) map[string]map[string]string {
rv := map[string]map[string]string{}
- if b.VBucketServerMap.ServerList == nil {
+ vsm := b.VBServerMap()
+ if vsm.ServerList == nil {
return rv
}
// Go grab all the things at once.
- todo := len(b.VBucketServerMap.ServerList)
+ todo := len(vsm.ServerList)
ch := make(chan gathered_stats, todo)
- for offset := range b.VBucketServerMap.ServerList {
+ for offset := range vsm.ServerList {
go getStatsParallel(b, offset, which, ch)
}
// Gather the results
- for i := 0; i < len(b.VBucketServerMap.ServerList); i++ {
+ for i := 0; i < len(vsm.ServerList); i++ {
g := <-ch
if len(g.vals) > 0 {
rv[g.sn] = g.vals
@@ -148,7 +150,7 @@ func (b *Bucket) doBulkGet(vb uint16, keys []string,
attempts := 0
done := false
for attempts < MaxBulkRetries && !done {
- masterId := b.VBucketServerMap.VBucketMap[vb][0]
+ masterId := b.VBServerMap().VBucketMap[vb][0]
attempts++
// This stack frame exists to ensure we can clean up
View
@@ -35,11 +35,11 @@ func doOps(b *couchbase.Bucket) {
}
func exploreBucket(bucket *couchbase.Bucket) {
- fmt.Printf(" %v uses %s\n", bucket.Name,
- bucket.VBucketServerMap.HashAlgorithm)
- for pos, server := range bucket.VBucketServerMap.ServerList {
+ vbm := bucket.VBServerMap()
+ fmt.Printf(" %v uses %s\n", bucket.Name, vbm.HashAlgorithm)
+ for pos, server := range vbm.ServerList {
vbs := make([]string, 0, 1024)
- for vb, a := range bucket.VBucketServerMap.VBucketMap {
+ for vb, a := range vbm.VBucketMap {
if a[0] == pos {
vbs = append(vbs, strconv.Itoa(vb))
}
View
@@ -13,6 +13,8 @@ import (
"runtime"
"sort"
"strings"
+ "sync/atomic"
+ "unsafe"
)
// The HTTP Client To Use
@@ -73,6 +75,13 @@ type Pool struct {
client Client
}
+type VBucketServerMap struct {
+ HashAlgorithm string `json:"hashAlgorithm"`
+ NumReplicas int `json:"numReplicas"`
+ ServerList []string `json:"serverList"`
+ VBucketMap [][]int `json:"vBucketMap"`
+}
+
// An individual bucket. Herein lives the most useful stuff.
type Bucket struct {
AuthType string `json:"authType"`
@@ -81,7 +90,6 @@ type Bucket struct {
Type string `json:"bucketType"`
Name string `json:"name"`
NodeLocator string `json:"nodeLocator"`
- Nodes []Node `json:"nodes"`
Quota map[string]float64 `json:"quota,omitempty"`
Replicas int `json:"replicaNumber"`
Password string `json:"saslPassword"`
@@ -92,22 +100,46 @@ type Bucket struct {
DDocs struct {
URI string `json:"uri"`
} `json:"ddocs,omitempty"`
- VBucketServerMap struct {
- HashAlgorithm string `json:"hashAlgorithm"`
- NumReplicas int `json:"numReplicas"`
- ServerList []string `json:"serverList"`
- VBucketMap [][]int `json:"vBucketMap"`
- } `json:"vBucketServerMap"`
BasicStats map[string]interface{} `json:"basicStats,omitempty"`
Controllers map[string]interface{} `json:"controllers,omitempty"`
- pool *Pool
- connPools []*connectionPool
- commonSufix string
+ // These are used for JSON IO, but isn't used for processing
+ // since it needs to be swapped out safely.
+ VBSMJson VBucketServerMap `json:"vBucketServerMap"`
+ NodesJson []Node `json:"nodes"`
+
+ pool *Pool
+ connPools unsafe.Pointer // *[]*connectionPool
+ vBucketServerMap unsafe.Pointer // *VBucketServerMap
+ nodeList unsafe.Pointer // *[]Node
+ commonSufix string
+}
+
+// Get the current vbucket server map
+func (b Bucket) VBServerMap() *VBucketServerMap {
+ return (*VBucketServerMap)(atomic.LoadPointer(&b.vBucketServerMap))
+}
+
+func (b Bucket) Nodes() []Node {
+ return *(*[]Node)(atomic.LoadPointer(&b.nodeList))
}
func (b Bucket) getConnPools() []*connectionPool {
- return b.connPools
+ return *(*[]*connectionPool)(atomic.LoadPointer(&b.connPools))
+}
+
+func (b *Bucket) replaceConnPools(with []*connectionPool) {
+ for {
+ old := atomic.LoadPointer(&b.connPools)
+ if atomic.CompareAndSwapPointer(&b.connPools, old, unsafe.Pointer(&with)) {
+ if old != nil {
+ for _, pool := range *(*[]*connectionPool)(old) {
+ pool.Close()
+ }
+ }
+ return
+ }
+ }
}
func (b Bucket) getConnPool(i int) *connectionPool {
@@ -130,16 +162,17 @@ func (b Bucket) authHandler() (ah AuthHandler) {
// Get the (sorted) list of memcached node addresses (hostname:port).
func (b Bucket) NodeAddresses() []string {
- rv := make([]string, len(b.VBucketServerMap.ServerList))
- copy(rv, b.VBucketServerMap.ServerList)
+ vsm := b.VBServerMap()
+ rv := make([]string, len(vsm.ServerList))
+ copy(rv, vsm.ServerList)
sort.Strings(rv)
return rv
}
// Get the longest common suffix of all host:port strings in the node list.
func (b Bucket) CommonAddressSuffix() string {
input := []string{}
- for _, n := range b.Nodes {
+ for _, n := range b.Nodes() {
input = append(input, n.Hostname)
}
return FindCommonSuffix(input)
@@ -235,16 +268,20 @@ func Connect(baseU string) (Client, error) {
func (b *Bucket) refresh() error {
pool := b.pool
- err := pool.client.parseURLResponse(b.URI, b)
+ tmpb := &Bucket{}
+ err := pool.client.parseURLResponse(b.URI, tmpb)
if err != nil {
return err
}
- b.pool = pool
- for i := range b.connPools {
- b.connPools[i] = newConnectionPool(
- b.VBucketServerMap.ServerList[i],
+ newcps := make([]*connectionPool, len(b.VBSMJson.ServerList))
+ for i := range newcps {
+ newcps[i] = newConnectionPool(
+ tmpb.VBSMJson.ServerList[i],
b.authHandler(), PoolSize, PoolOverflow)
}
+ b.replaceConnPools(newcps)
+ atomic.StorePointer(&b.vBucketServerMap, unsafe.Pointer(&b.VBSMJson))
+ atomic.StorePointer(&b.nodeList, unsafe.Pointer(&b.NodesJson))
return nil
}
@@ -258,7 +295,8 @@ func (p *Pool) refresh() (err error) {
}
for _, b := range buckets {
b.pool = p
- b.connPools = make([]*connectionPool, len(b.VBucketServerMap.ServerList))
+ b.nodeList = unsafe.Pointer(&b.NodesJson)
+ b.replaceConnPools(make([]*connectionPool, len(b.VBSMJson.ServerList)))
p.BucketMap[b.Name] = b
}
@@ -288,7 +326,7 @@ 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.connPools != nil {
- for _, c := range b.connPools {
+ for _, c := range b.getConnPools() {
if c != nil {
c.Close()
}
View
@@ -3,6 +3,7 @@ package couchbase
import (
"encoding/json"
"testing"
+ "unsafe"
)
var samplePools = `{
@@ -288,24 +289,25 @@ func TestPool(t *testing.T) {
}
func TestCommonAddressSuffixEmpty(t *testing.T) {
- b := Bucket{}
+ b := Bucket{nodeList: mkNL([]Node{})}
assert(t, "empty", "", b.CommonAddressSuffix())
}
func TestCommonAddressSuffixUncommon(t *testing.T) {
- b := Bucket{}
- b.VBucketServerMap.ServerList = []string{"somestring", "unrelated"}
+ b := Bucket{vBucketServerMap: unsafe.Pointer(&VBucketServerMap{
+ ServerList: []string{"somestring", "unrelated"}}),
+ nodeList: mkNL([]Node{}),
+ }
assert(t, "shouldn't match", "", b.CommonAddressSuffix())
}
func TestCommonAddressSuffixCommon(t *testing.T) {
- b := Bucket{}
- b.Nodes = []Node{
+ b := Bucket{nodeList: unsafe.Pointer(&[]Node{
{Hostname: "server1.example.com:11210"},
{Hostname: "server2.example.com:11210"},
{Hostname: "server3.example.com:11210"},
{Hostname: "server4.example.com:11210"},
- }
+ })}
assert(t, "useful suffix", ".example.com:11210",
b.CommonAddressSuffix())
}
View
@@ -74,5 +74,6 @@ func (b *Bucket) VBHash(key string) uint32 {
for x := 0; x < len(key); x++ {
crc = (crc >> 8) ^ crc32tab[(uint64(crc)^uint64(key[x]))&0xff]
}
- return ((^crc) >> 16) & 0x7fff & (uint32(len(b.VBucketServerMap.VBucketMap)) - 1)
+ vbm := b.VBServerMap()
+ return ((^crc) >> 16) & 0x7fff & (uint32(len(vbm.VBucketMap)) - 1)
}
View
@@ -2,11 +2,13 @@ package couchbase
import (
"testing"
+ "unsafe"
)
func testBucket() Bucket {
- b := Bucket{}
- b.VBucketServerMap.VBucketMap = make([][]int, 256)
+ b := Bucket{vBucketServerMap: unsafe.Pointer(&VBucketServerMap{
+ VBucketMap: make([][]int, 256),
+ })}
return b
}
View
@@ -32,11 +32,12 @@ type ViewResult struct {
}
func (b *Bucket) randomBaseURL() (*url.URL, error) {
- if len(b.Nodes) == 0 {
+ if len(b.Nodes()) == 0 {
return nil, errors.New("no couch rest URLs")
}
- nodeNo := rand.Intn(len(b.Nodes))
- node := b.Nodes[nodeNo]
+ nodes := b.Nodes()
+ nodeNo := rand.Intn(len(nodes))
+ node := nodes[nodeNo]
if node.CouchAPIBase == "" {
// Probably in "warmup" state
return nil, fmt.Errorf("Bucket is in %q state, not ready for view queries", node.Status)
View
@@ -2,32 +2,30 @@ package couchbase
import (
"testing"
+ "unsafe"
)
-func TestViewURL(t *testing.T) {
- b := Bucket{}
- // No URLs
- v, err := b.ViewURL("a", "b", nil)
- if err == nil {
- t.Errorf("Expected error on empty bucket, got %v", v)
- }
+func mkNL(in []Node) unsafe.Pointer {
+ return unsafe.Pointer(&in)
+}
+func TestViewURL(t *testing.T) {
// Missing URL
- b = Bucket{Nodes: []Node{{}}}
- v, err = b.ViewURL("a", "b", nil)
+ b := Bucket{nodeList: mkNL([]Node{{}})}
+ v, err := b.ViewURL("a", "b", nil)
if err == nil {
t.Errorf("Expected error on missing URL, got %v", v)
}
// Invalidish URL
- b = Bucket{Nodes: []Node{{CouchAPIBase: "::gopher:://localhost:80x92/"}}}
+ b = Bucket{nodeList: mkNL([]Node{{CouchAPIBase: "::gopher:://localhost:80x92/"}})}
v, err = b.ViewURL("a", "b", nil)
if err == nil {
t.Errorf("Expected error on broken URL, got %v", v)
}
// Unmarshallable parameter
- b = Bucket{Nodes: []Node{{CouchAPIBase: "http:://localhost:8092/"}}}
+ b = Bucket{nodeList: mkNL([]Node{{CouchAPIBase: "http:://localhost:8092/"}})}
v, err = b.ViewURL("a", "b",
map[string]interface{}{"ch": make(chan bool)})
if err == nil {
@@ -59,7 +57,8 @@ func TestViewURL(t *testing.T) {
{"", "_all_docs", nil, "/x/_all_docs", map[string]string{}},
}
- b = Bucket{Name: "x", Nodes: []Node{{CouchAPIBase: "http://localhost:8092/"}}}
+ b = Bucket{Name: "x",
+ nodeList: mkNL([]Node{{CouchAPIBase: "http://localhost:8092/"}})}
for _, test := range tests {
us, err := b.ViewURL(test.ddoc, test.name, test.params)
if err != nil {

0 comments on commit 9a55f93

Please sign in to comment.