scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into rbf-fix-direct-write-corruption

Browse source
This commit is contained in:
Ben Johnson 2020-09-16 09:55:26 -06:00 • committed by GitHub
commit 6918fe2f60
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 290 additions and 328 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

145
dbshard.go
View file

@ -22,7 +22,6 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"github.com/pkg/errors"
)
@ -192,6 +191,11 @@ func (dbs *DBShard) DeleteDBPath() (err error) {
return
}
type flatkey struct {
index string
shard uint64
}
type DBPerShard struct {
Mu sync.Mutex
@ -201,12 +205,13 @@ type DBPerShard struct {
// just flat, not buried within the Node heirarchy.
// Easily see how many we have.
Flatmap map[*DBShard]struct{}
Flatmap map[flatkey]*DBShard
types []txtype
hasRoaring bool
txf *TxFactory
txf *TxFactory
holder *Holder
// which of our types is not-roaring, since
// roaring doesn't keep a list of open Tx sn.
@ -215,13 +220,13 @@ type DBPerShard struct {
}
// HasData returns true if the database has at least one key.
// For roaring it returns the number of fragments stored.
// For roaring it returns true if we a fragment stored.
// The `which` argument is the index into the per.W slice. 0 for blue, 1 for green.
// If you pass 1, be sure you have a blue-green configuration.
func (per *DBPerShard) HasData(which int) (hasData bool, err error) {
// has to aggregate across all available DBShard for each index and shard.
for v := range per.Flatmap {
for _, v := range per.Flatmap {
hasData, err = v.W[which].HasData()
if err != nil {
return
@ -234,13 +239,32 @@ func (per *DBPerShard) HasData(which int) (hasData bool, err error) {
}
func (per *DBPerShard) ListOpenString() (r string) {
for v := range per.Flatmap {
for _, v := range per.Flatmap {
r += v.HolderPath + " -> " + v.W[per.useOpenList].OpenListString() + "\n"
}
return
}
func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string) (d *DBPerShard) {
func (per *DBPerShard) LoadExistingDBs() (err error) {
idxs := per.holder.Indexes()
for _, idx := range idxs {
sos, err := per.txf.GetShardsForIndex(idx, "", true)
if err != nil {
return err
}
for _, shard := range sos {
_, err := per.GetDBShard(idx.name, shard, idx)
if err != nil {
return errors.Wrap(err, "DBPerShard.LoadExistingDBs GetDBShard()")
}
}
}
return
}
func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string, holder *Holder) (d *DBPerShard) {
useOpenList := 0
hasRoaring := false
@ -261,8 +285,9 @@ func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string) (d *DBPerS
d = &DBPerShard{
types: types,
HolderDir: holderDir,
holder: holder,
dbh: NewDBHolder(),
Flatmap: make(map[*DBShard]struct{}),
Flatmap: make(map[flatkey]*DBShard),
txf: txf,
useOpenList: useOpenList,
hasRoaring: hasRoaring,
@ -327,7 +352,9 @@ func (per *DBPerShard) DeleteFragment(index, field, view string, shard uint64, f
idx := per.txf.holder.Index(index)
dbs, err := per.GetDBShard(index, shard, idx)
panicOn(err)
if err != nil {
return err
}
return dbs.DeleteFragment(index, field, view, shard, frag)
}
@ -357,6 +384,7 @@ func (dbs *DBShard) DumpAll() {
func (per *DBPerShard) DumpAll() {
per.Mu.Lock()
defer per.Mu.Unlock()
found1 := false
for _, dbi := range per.dbh.Index {
for _, dbs := range dbi.Shard {
@ -447,7 +475,7 @@ func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *
w.SetHolder(h)
dbs.Open = true
if w != nil && len(dbs.W) == 0 {
per.Flatmap[dbs] = struct{}{}
per.Flatmap[flatkey{index: index, shard: shard}] = dbs
}
dbs.W = append(dbs.W, w)
}
@ -464,7 +492,7 @@ func (per *DBPerShard) Del(dbs *DBShard) (err error) {
return
}
panicOn(dbs.DeleteDBPath())
delete(per.Flatmap, dbs)
delete(per.Flatmap, flatkey{index: dbs.Index, shard: dbs.Shard})
// delete from the heirarchy
delete(dbs.ParentDBIndex.Shard, dbs.Shard)
@ -485,12 +513,14 @@ func (per *DBPerShard) Close() (err error) {
}
// DBPerShardGetShardsForIndex returns the shards for idx.
func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string) (sliceOfShards []uint64, err error) {
// If requireData, we open the database and see that it has a key, rather
// than assume that the database file presence is enough.
func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requireData bool) (sliceOfShards []uint64, err error) {
var shards [][]uint64
for _, ty := range f.types {
var slc []uint64
slc, err = f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath)
slc, err = f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath, requireData)
if err != nil {
return
}
@ -500,32 +530,20 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string) (slice
if n != 1 && n != 2 {
panic(fmt.Sprintf("internal error. only green or blue/green supported. we see types len %v", n))
}
if !f.blueGreenOff && n == 2 {
// this is a blue green check which cannot live inside Tx because we don't know the shard yet.
// Therefore it has to be above Tx, since we are getting all the shards to choose from here.
//
// But, we still want to check for blue-green consistency. In fact, this was written
// in response to an issue with balancing/re-balancing shards being different
// over the cluster of nodes between blue and green.
b := sliceToMap(shards[0])
g := sliceToMap(shards[1])
blueMinusGreenDiff := mapDiff(b, g)
greenMinusBlueDiff := mapDiff(g, b)
if len(blueMinusGreenDiff) == 0 && len(greenMinusBlueDiff) == 0 {
// ok
} else {
vv("blue[%v] and green[%v] have different shards for index '%v': blueMinusGreenDiff: %v, greenMinusBlueDiff: %v; blueShards='%v', greenShards='%v'; idx.path='%v'", f.types[0].String(), f.types[1].String(), idx.name, blueMinusGreenDiff, greenMinusBlueDiff, asInts(shards[0]), asInts(shards[1]), idx.path)
panic(fmt.Sprintf("blue[%v] and green[%v] have different shards for index '%v': blueMinusGreenDiff: %v, greenMinusBlueDiff: %v; blueShards='%v', greenShards='%v'; idx.path='%v'", f.types[0].String(), f.types[1].String(), idx.name, blueMinusGreenDiff, greenMinusBlueDiff, asInts(shards[0]), asInts(shards[1]), idx.path))
}
}
// Note: we don't actually know when the blue call and when the green call comes
// through here. So if we are deleting a shard, we will see a difference earlier
// in one than the other. TestAPI_ClearFlagForImportAndImportValues for example.
// Hence we cannot do a blue-green check here for matching shards.
// If we are populating blue from green, it does matter that we return green.
return shards[n-1], nil
}
// if roaringViewPath is "" then for ty == roaringTxn we go to disk to discover
// all the view paths under idx for type ty.
func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string) (sliceOfShards []uint64, err error) {
// requireData means open the database file and verify that at least one key is set.
func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (sliceOfShards []uint64, err error) {
if ty == roaringTxn {
rx := &RoaringTx{
@ -549,7 +567,6 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
return rx.SliceOfShards("", "", "", roaringViewPath)
}
requiredSuffix := ty.FileSuffix()
//path := idx.Path()
path := per.prefixForType(idx, ty)
ignoreEmpty := false
@ -558,7 +575,9 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
panicOn(err)
for _, nm := range dbf {
base := filepath.Base(nm)
splt := strings.Split(base, requiredSuffix)
if len(splt) != 2 {
panic(fmt.Sprintf("should have 2 parts: nm='%v', base(nm)='%v'; requiredSuffix='%v'", nm, base, requiredSuffix))
@ -569,21 +588,57 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
if !strings.HasPrefix(prefix, shardPrefix) {
continue
}
// Parse filename into integer.
shard, err := strconv.ParseUint(prefix[lenOfShardPrefix:], 10, 64)
if err != nil {
panicOn(err)
continue
}
sliceOfShards = append(sliceOfShards, shard)
// exclude those without data?
hasData := false
if requireData {
hasData, err = per.TypedIndexShardHasData(ty, idx, shard)
if err != nil {
return nil, err
}
if hasData {
sliceOfShards = append(sliceOfShards, shard)
}
} else {
// file presence is enough
sliceOfShards = append(sliceOfShards, shard)
}
}
return
}
func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) {
whichty := 0
if len(per.types) == 2 {
if ty == per.types[1] {
whichty = 1
}
}
if ty != per.types[whichty] {
return
}
// make the dbs if it doesn't get exist
dbs, err := per.GetDBShard(idx.name, shard, idx)
if err != nil {
return false, errors.Wrap(err, fmt.Sprintf("DBPerShard.TypedIndexShardHasData() "+
"per.GetDBShard(index='%v', shard='%v', ty='%v')", idx.name, shard, ty.String()))
}
return dbs.W[whichty].HasData()
}
func listDirUnderDir(root string, includeRoot bool, requiredSuffix string, ignoreEmpty bool) (files []string, err error) {
//vv("listDirUnderDir(root ='%v', suffix='%v')", root, requiredSuffix)
if !dirExists(root) {
//vv("warning: listFilesUnderDir error: root directory '%v' not found", root)
//return nil, fmt.Errorf("listFilesUnderDir error: root directory '%v' not found", root)
return
}
n := len(root) + 1
@ -649,9 +704,10 @@ func listDirUnderDir(root string, includeRoot bool, requiredSuffix string, ignor
// populateBlueFromGreen().
//
func (dbs *DBShard) populateBlueFromGreen() (err error) {
n := len(dbs.W)
if n != 2 {
panic(fmt.Sprintf("copyGreenToBlue did not find 2 open DBs: have %v", n))
panic(fmt.Sprintf("populateBlueFromGreen did not find 2 open DBs: have %v", n))
}
dest := dbs.W[0] // blue
@ -674,7 +730,12 @@ func (dbs *DBShard) populateBlueFromGreen() (err error) {
view := vw.name
citer, _, err := readtx.ContainerIterator(dbs.Index, field, view, dbs.Shard, 0)
if err != nil {
return errors.Wrap(err, "DBShard.copyGreenToBlue readtx.ContainerIterator")
// might be an empty fragment. If so, let's not freak out.
if strings.HasPrefix(err.Error(), "fragment not found") {
continue
} else {
return errors.Wrap(err, "DBShard.populateBlueFromGreen readtx.ContainerIterator")
}
}
for citer.Next() {
@ -682,7 +743,7 @@ func (dbs *DBShard) populateBlueFromGreen() (err error) {
err := writetx.PutContainer(dbs.Index, field, view, dbs.Shard, ckey, rc)
if err != nil {
citer.Close()
return errors.Wrap(err, "DBShard.copyGreenToBlue writetx.PutContainer")
return errors.Wrap(err, "DBShard.populateBlueFromGreen writetx.PutContainer")
}
}
citer.Close()
@ -696,7 +757,7 @@ func (dbs *DBShard) populateBlueFromGreen() (err error) {
}
// verifyBlueEqualsGreen checks that blue and green are identical.
func (dbs *DBShard) verifyBlueEqualsGreen(numCtVerified *int64) (err error) {
func (dbs *DBShard) verifyBlueEqualsGreen() (err error) {
n := len(dbs.W)
if n != 2 {
@ -773,8 +834,6 @@ func (dbs *DBShard) verifyBlueEqualsGreen(numCtVerified *int64) (err error) {
"shard: '%v' ckey: '%v' nHotGreen= %v nHotBlue= %v ; BitwiseCompare response: '%v'",
dbs.Index, field, view, dbs.Shard, greenCkey, nGreen, nBlue, err))
}
atomic.AddInt64(numCtVerified, 1)
}
if bCiter.Next() {
blueCkey, _ := bCiter.Value()

151
dbshard_internal_test.go
View file

@ -21,6 +21,8 @@ import (
"path/filepath"
"strings"
"testing"
"github.com/pilosa/pilosa/v2/rbf"
)
// Shard per db evaluation
@ -73,19 +75,22 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!
for _, src := range []string{"lmdb", "roaring", "badger", "rbf"} {
makeSampleRoaringDir(tmpdir, src, 0)
os.Setenv("PILOSA_TXSRC", src)
// must make Holder AFTER setting src.
holder := NewHolder(tmpdir, nil)
makeSampleRoaringDir(tmpdir, src, 1, holder)
idx, err := NewIndex(holder, tmpdir, "rick")
panicOn(err)
estd := "rick/_exists/views/standard"
std := "rick/f/views/standard"
sos, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std)
sos, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false)
panicOn(err)
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
if !inSlice(sos, shard) {
panic(fmt.Sprintf("missing shard=%v from sos='%#v'", shard, sos))
@ -93,7 +98,7 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
}
if src == "roaring" {
// check estd too
sos, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd)
sos, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false)
panicOn(err)
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
if !inSlice(sos, shard) {
@ -137,67 +142,63 @@ rick/_exists/views/standard/fragments/219
rick/_exists/views/standard/fragments/223
`,
"lmdb": `
rick.index.txstores@@@/store-lmdb@@/shard.0219-lmdb@/data.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0219-lmdb@/lock.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0093-lmdb@/data.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0093-lmdb@/lock.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0223-lmdb@/data.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0223-lmdb@/lock.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0215-lmdb@/data.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0215-lmdb@/lock.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0217-lmdb@/data.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0217-lmdb@/lock.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0221-lmdb@/data.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0221-lmdb@/lock.mdb
rick.index.txstores@@@/store-lmdb@@/shard.0093-lmdb@
rick.index.txstores@@@/store-lmdb@@/shard.0215-lmdb@
rick.index.txstores@@@/store-lmdb@@/shard.0217-lmdb@
rick.index.txstores@@@/store-lmdb@@/shard.0219-lmdb@
rick.index.txstores@@@/store-lmdb@@/shard.0221-lmdb@
rick.index.txstores@@@/store-lmdb@@/shard.0223-lmdb@
`,
"badger": `
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/000000.vlog
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/KEYREGISTRY
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/MANIFEST
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/LOCK
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/000000.vlog
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/KEYREGISTRY
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/MANIFEST
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/LOCK
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/000000.vlog
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/KEYREGISTRY
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/MANIFEST
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/LOCK
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/000000.vlog
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/KEYREGISTRY
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/MANIFEST
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/LOCK
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/000000.vlog
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/KEYREGISTRY
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/MANIFEST
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/LOCK
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/000000.vlog
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/KEYREGISTRY
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/MANIFEST
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/LOCK
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@
`,
"rbf": `
rick.index.txstores@@@/store-rbfdb@@/shard.0223-rbfdb@/wal/0000000000000001.wal
rick.index.txstores@@@/store-rbfdb@@/shard.0223-rbfdb@/data
rick.index.txstores@@@/store-rbfdb@@/shard.0093-rbfdb@/wal/0000000000000001.wal
rick.index.txstores@@@/store-rbfdb@@/shard.0093-rbfdb@/data
rick.index.txstores@@@/store-rbfdb@@/shard.0217-rbfdb@/wal/0000000000000001.wal
rick.index.txstores@@@/store-rbfdb@@/shard.0217-rbfdb@/data
rick.index.txstores@@@/store-rbfdb@@/shard.0215-rbfdb@/wal/0000000000000001.wal
rick.index.txstores@@@/store-rbfdb@@/shard.0215-rbfdb@/data
rick.index.txstores@@@/store-rbfdb@@/shard.0221-rbfdb@/wal/0000000000000001.wal
rick.index.txstores@@@/store-rbfdb@@/shard.0221-rbfdb@/data
rick.index.txstores@@@/store-rbfdb@@/shard.0219-rbfdb@/wal/0000000000000001.wal
rick.index.txstores@@@/store-rbfdb@@/shard.0219-rbfdb@/data
rick.index.txstores@@@/store-rbfdb@@/shard.0093-rbfdb@
rick.index.txstores@@@/store-rbfdb@@/shard.0215-rbfdb@
rick.index.txstores@@@/store-rbfdb@@/shard.0217-rbfdb@
rick.index.txstores@@@/store-rbfdb@@/shard.0219-rbfdb@
rick.index.txstores@@@/store-rbfdb@@/shard.0221-rbfdb@
rick.index.txstores@@@/store-rbfdb@@/shard.0223-rbfdb@
`,
}
func makeSampleRoaringDir(root, txsrc string, minBytes int) {
func makeSampleRoaringDir(root, txsrc string, minBytes int, h *Holder) {
index := "rick"
shards := []uint64{0, 93, 215, 217, 219, 221, 223}
fns := strings.Split(sampleRoaringDirList[txsrc], "\n")
for _, fn := range fns {
for i, fn := range fns {
if fn == "" {
continue
}
var shard uint64
if txsrc != "roaring" {
// only have shards for the non-roaring
shard = shards[i]
}
switch txsrc {
case "lmdb":
makeLMDBtestDB(root+sep+fn, h, shard)
// also have to make the DBShard in our in-memory tree,
// or else the search won't find it because
// DBPerShard won't know anything about it.
helperCreateDBShard(h, index, shard)
continue
case "badger":
makeBadgertestDB(root+sep+fn, h, shard)
helperCreateDBShard(h, index, shard)
continue
case "rbf":
makeRBFtestDB(root+sep+fn, h, shard)
helperCreateDBShard(h, index, shard)
continue
}
path := root + sep + filepath.Dir(fn)
panicOn(os.MkdirAll(path, 0755))
fd, err := os.Create(root + sep + fn)
@ -209,3 +210,47 @@ func makeSampleRoaringDir(root, txsrc string, minBytes int) {
fd.Close()
}
}
func helperCreateDBShard(h *Holder, index string, shard uint64) {
idx, err := h.CreateIndexIfNotExists(index, IndexOptions{})
panicOn(err)
dbs, err := h.txf.dbPerShard.GetDBShard(index, shard, idx)
panicOn(err)
_ = dbs
}
func makeLMDBtestDB(path string, h *Holder, shard uint64) {
i := uint64(1)
w, _ := mustOpenEmptyLMDBWrapper(path)
LMDBMustSetBitvalue(w, "index", "field", "view", shard, i)
w.Close()
}
func makeBadgertestDB(path string, h *Holder, shard uint64) {
i := uint64(1)
w, _ := mustOpenEmptyBadgerWrapper(path)
badgerDBMustSetBitvalue(w, "index", "field", "view", shard, i)
w.Close()
}
func makeRBFtestDB(path string, h *Holder, shard uint64) {
i := uint64(1)
db := rbf.NewDB(path)
err := db.Open()
panicOn(err)
defer db.Close()
tx, err := db.Begin(true)
panicOn(err)
err = tx.CreateBitmap("x")
panicOn(err)
_, err = tx.Add("x", i)
panicOn(err)
err = tx.Commit()
panicOn(err)
}

1
go.mod
View file

@ -12,7 +12,6 @@ require (
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect
github.com/davecgh/go-spew v1.1.1
github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200709123515-8e896a7af361
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311
github.com/glycerine/lmdb-go v1.9.32
github.com/go-ole/go-ole v1.2.4 // indirect
github.com/gogo/protobuf v1.2.1

135
holdbkg.go
View file

@ -1,135 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
import (
"github.com/glycerine/idem"
)
// holderBackgroundGoro avoids the deadlock versus race dilmena
// when creating a new index. The troubles were in having
// the field/view code try to inform the Holder of the indexes
// from a platoon of worker goroutines and tests; see index.go:273 inside
// Index.openFields(). There calling i.holder.addIndex(i) instead
// of locking cleans things up alot. There's no need anymore
// to track which goroutines are holding the Holder's
// mu lock.
type holderBackgroundGoro struct {
h *Holder
halt *idem.Halter
reqIndexCh chan *indexReq
setIdxCh chan *Index
getAllCh chan *indexReq
delIdxCh chan string
}
// indexReq is used to ask the holderBackgrounGoro
// for index information.
type indexReq struct {
// basic request to map an index name to *Index
index string
idx *Index
// double duty as a getAll indexes request
getAll bool
all []*Index
done chan struct{}
}
func newIndexReq(index string) *indexReq {
return &indexReq{
index: index,
done: make(chan struct{}),
}
}
func (b *holderBackgroundGoro) index(index string) *Index {
if b == nil {
// utils_internal_test.go:448 from
// TestCluster_ResizeStates/Multiple_nodes,_with_data
// wants to pass a nil b/nil Holder. sigh. don't panic.
return nil
}
req := newIndexReq(index)
b.reqIndexCh <- req
<-req.done
return req.idx
}
func (b *holderBackgroundGoro) isClosed() bool {
select {
case <-b.halt.Done.Chan:
return true
default:
return false
}
}
func (b *holderBackgroundGoro) start() {
go func() {
defer b.halt.Done.Close()
for {
select {
case <-b.halt.ReqStop.Chan:
return
case r := <-b.reqIndexCh:
r.idx = b.h.indexesOwnedByBkgr[r.index]
close(r.done)
case idx := <-b.setIdxCh:
b.h.indexesOwnedByBkgr[idx.Name()] = idx
// atomic operation, client doesn't need to wait on done,
// so don't take the time to do another channel, just leave done open.
case target := <-b.delIdxCh:
delete(b.h.indexesOwnedByBkgr, target)
// atomic operation, client doesn't need to wait on done,
// so don't take the time to do another channel, just leave done open.
case r := <-b.getAllCh:
r.all = make([]*Index, 0, len(b.h.indexesOwnedByBkgr))
for _, index := range b.h.indexesOwnedByBkgr {
r.all = append(r.all, index)
}
close(r.done)
}
}
}()
}
func (h *Holder) stopBkgr() {
h.mu.Lock()
defer h.mu.Unlock()
if h.bkgr != nil {
h.bkgr.halt.ReqStop.Close()
<-h.bkgr.halt.Done.Chan
}
}
func (h *Holder) newHolderBackgroundGoro() (b *holderBackgroundGoro) {
b = &holderBackgroundGoro{
h: h,
halt: idem.NewHalter(),
reqIndexCh: make(chan *indexReq),
setIdxCh: make(chan *Index),
getAllCh: make(chan *indexReq),
delIdxCh: make(chan string),
}
b.start()
return b
}

99
holder.go
View file

@ -117,31 +117,10 @@ type Holder struct {
txf *TxFactory
bkgr *holderBackgroundGoro
// indexesOwnedByBkgr is owned by h.bkgr. Do not touch.
// Only the owning holdbkg.go goroutine should query or
// modify the indexesOwnedByBkgr map. indexesOwnedByBkgr replaces
// the old indexes map which was the source of deadlock vs
// race issues with a dedicated goroutine
// associated with the Holder.
//
// Normally this map would live inside the holderBackgroundGoro
// struct, but tests expect the map to
// persist cross Holder ReOpens(), so it still lives here --
// since a ReOpen will kill and restart the goroutine and
// lose its state.
//
// Again, do not touch this directly. Use these methods instead:
//
// h.Index() // index name -> *Index
// h.Indexes() // copy of the full list of *Indexes
// h.addIndex() // add one *Index
// h.deleteIndex() // delete one *Index
//
indexesOwnedByBkgr map[string]*Index
numCtBlueGreenVerified int64
// a separate lock out for indexes, to avoid the deadlock/race dilema
// on holding mu.
imu sync.RWMutex
indexes map[string]*Index
}
// HolderOpts holds information about the holder which other things might want
@ -235,8 +214,6 @@ func DefaultHolderConfig() *HolderConfig {
}
// NewHolder returns a new instance of Holder for the given path.
// It starts the bkgr background goroutine that provides
// exclusive access to the indexesOwnedByBkgr map.
func NewHolder(path string, cfg *HolderConfig) *Holder {
if cfg == nil {
cfg = DefaultHolderConfig()
@ -274,13 +251,13 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
path: path,
indexesOwnedByBkgr: make(map[string]*Index),
indexes: make(map[string]*Index),
}
h.bkgr = h.newHolderBackgroundGoro()
txf, err := NewTxFactory(cfg.Txsrc, path, h)
panicOn(err)
h.txf = txf
h.txf.blueGreenOffIfRunningBlueGreen()
_ = testhook.Created(h.Auditor, h, nil)
return h
@ -571,10 +548,6 @@ func (h *Holder) Open() error {
h.opening = true
defer func() { h.opening = false }()
if h.bkgr == nil || h.bkgr.isClosed() {
h.bkgr = h.newHolderBackgroundGoro()
}
if h.txf == nil {
txf, err := NewTxFactory(h.cfg.Txsrc, h.path, h)
if err != nil {
@ -673,6 +646,10 @@ func (h *Holder) Open() error {
_ = testhook.Opened(h.Auditor, h, nil)
if err := h.txf.Open(); err != nil {
return errors.Wrap(err, "Holder.Open h.txf.Open()")
}
// under blue_green, we must sync blue from green before we turn on checking.
if err := h.txf.green2blue(h); err != nil {
return errors.Wrap(err, "Holder.Open h.txf.green2blue(h)")
@ -728,7 +705,6 @@ func (h *Holder) Close() error {
if h == nil {
return nil
}
defer h.stopBkgr()
if globalUseStatTx {
fmt.Printf("%v\n", globalCallStats.report())
@ -977,25 +953,24 @@ func (h *Holder) HolderPathFromIndexPath(indexPath, indexName string) string {
// Index returns the index by name.
func (h *Holder) Index(name string) (idx *Index) {
idx = h.bkgr.index(name)
h.imu.RLock()
idx = h.indexes[name]
h.imu.RUnlock()
return
}
// Indexes returns a list of all indexes in the holder.
func (h *Holder) Indexes() []*Index {
// utils_internal_test.go:345 func (t *ClusterCluster) Close() error
// wants to close an un-Open()-ed Holder.
// So I guess we won't panic here.
if h.bkgr == nil || h.bkgr.isClosed() {
return nil
h.imu.RLock()
// sizing and copying has to be done under the lock to avoid
// a logical race with a deletion/addition to indexes.
cp := make([]*Index, 0, len(h.indexes))
for _, idx := range h.indexes {
cp = append(cp, idx)
}
req := newIndexReq("")
req.getAll = true
h.bkgr.getAllCh <- req
<-req.done
sort.Sort(indexSlice(req.all))
return req.all
h.imu.RUnlock()
sort.Sort(indexSlice(cp))
return cp
}
// CreateIndex creates an index.
@ -1107,7 +1082,9 @@ func (h *Holder) DeleteIndex(name string) error {
}
func (h *Holder) deleteIndex(index string) {
h.bkgr.delIdxCh <- index
h.imu.Lock()
delete(h.indexes, index)
h.imu.Unlock()
}
// Field returns the field for an index and name.
@ -1973,25 +1950,9 @@ func (h *Holder) Process(ctx context.Context, op HolderOperator) (err error) {
// used by Index.openFields(), enabling Tx / Txf by telling
// the holder about its own indexes.
func (h *Holder) addIndex(idx *Index) {
if idx == nil {
panic("cannot pass nil to addIndex")
}
if h == nil {
panic("cannot call addIndex on nil Holder")
}
if h.bkgr == nil || h.bkgr.isClosed() {
// ugh. TestCluster_ResizeStates/Multiple_nodes,_with_data test
// from cluster_internal_test.go:799
// via utils_internal_test.go:450
// gets here, with the h.mu already held.
// so we cannot panic and complain or we mess up that test.
// But really, we should not be calling addIndex() on
// Holder that has not be Open()-ed.
h.mu.Lock()
h.bkgr = h.newHolderBackgroundGoro()
h.mu.Unlock()
}
h.bkgr.setIdxCh <- idx
h.imu.Lock()
h.indexes[idx.name] = idx
h.imu.Unlock()
}
func (h *Holder) DumpAllShards() {
@ -2012,10 +1973,6 @@ func (h *Holder) BeginTx(writable bool, idx *Index, shard uint64) (Tx, error) {
return h.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}), nil
}
func (h *Holder) NumContainersBlueGreenVerified() int {
return int(h.numCtBlueGreenVerified)
}
func (h *Holder) HasRoaringData() (has bool, err error) {
idxs := h.Indexes()

3
rbf/cursorx.go
View file

@ -152,6 +152,9 @@ func (c *Cursor) CurrentPageType() int {
func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
if len(l.Data) == 0 {
return nil
}
orig := l.Data
var cpMaybe []byte
var mapped bool

43
txfactory.go
View file

@ -493,7 +493,7 @@ func MustTxsrcToTxtype(txsrc string) (types []txtype) {
// NewTxFactory always opens an existing database. If you
// want to a fresh database, os.RemoveAll on dir/name ahead of time.
// We always store files in a subdir of dir. If we are having one
// We always store files in a subdir of holderDir. If we are having one
// database or many can depend on name.
func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory, err error) {
types := MustTxsrcToTxtype(txsrc)
@ -506,11 +506,17 @@ func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory,
if len(types) == 2 {
f.blueGreenReg = newBlueGreenReg(types)
}
f.dbPerShard = f.NewDBPerShard(types, holderDir)
f.dbPerShard = f.NewDBPerShard(types, holderDir, holder)
return f, err
}
// Open should be called only once the index metadata is loaded
// from Holder.Open(), so we find all of our indexes.
func (f *TxFactory) Open() error {
return f.dbPerShard.LoadExistingDBs()
}
// Txo holds the transaction options
type Txo struct {
Write bool
@ -1175,6 +1181,7 @@ func (f *TxFactory) blueHasData() (hasData bool, err error) {
// Called by test Test_TxFactory_UpdateBlueFromGreen_OnStartup() in
// txfactory_internal_test.go as well.
//
// This is a noop if we aren't running under a blue_green PILOSA_TXSRC.
func (f *TxFactory) green2blue(holder *Holder) (err error) {
// Holder.Open will always call us, even without blue_green. Which is fine.
@ -1201,32 +1208,32 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) {
for _, idx := range idxs {
// scan directories
blueShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(blueDest, idx, "")
blueShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(blueDest, idx, "", false)
if err != nil {
return errors.Wrap(err, fmt.Sprintf("GetDBShard(index='%v') error fetching blueShards", idx.name))
}
//vv("from blueDest='%v', blueShards = '%#v'", blueDest, blueShards)
// scan directories
greenShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(greenSrc, idx, "")
greenShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(greenSrc, idx, "", true)
if err != nil {
return errors.Wrap(err, fmt.Sprintf("GetDBShard(index='%v') error fetching greenShards", idx.name))
}
//vv("from greenSrc='%v', greenShards = '%#v'", greenSrc, greenShards)
diff := f.shardSliceDiff(blueShards, greenShards)
if diff != "" {
return fmt.Errorf("blue[%v] and green[%v] have different shards for index '%v': %v", blueDest, greenSrc, idx.name, diff)
if verifyInsteadOfCopy {
diff := f.shardSliceDiff(blueShards, greenShards)
if diff != "" {
return fmt.Errorf("verifyInsteadOfCopy true, blue[%v]=%#v and green[%v]=%#v have different shards for index '%v': '%v'; stack=\n%v", blueDest, blueShards, greenSrc, greenShards, idx.name, diff, stack())
}
// can also check against meta data
shards := idx.AvailableShards(localOnly).Slice()
diff2 := f.shardSliceDiff(greenShards, shards)
if diff2 != "" {
return fmt.Errorf("green[%v] = '%#v' and meta data '%#v' have different shards for index '%v': %v", greenSrc, greenShards, shards, idx.name, diff2)
}
}
shards := idx.AvailableShards(localOnly).Slice()
diff2 := f.shardSliceDiff(blueShards, shards)
if diff2 != "" {
return fmt.Errorf("blue[%v] and meta data (from green[%v]?)have different shards for index '%v': %v", blueDest, greenSrc, idx.name, diff2)
}
for _, shard := range shards {
for _, shard := range greenShards {
dbs, err := f.dbPerShard.GetDBShard(idx.name, shard, idx)
if err != nil {
@ -1235,7 +1242,7 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) {
if verifyInsteadOfCopy {
// verify all containers
err = dbs.verifyBlueEqualsGreen(&holder.numCtBlueGreenVerified)
err = dbs.verifyBlueEqualsGreen()
if err != nil {
return errors.Wrap(err,
fmt.Sprintf("dbs.verifyBlueEqualsGreen(blue='%v', "+

39
txfactory_internal_test.go
View file

@ -112,12 +112,11 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) {
// and b) we have an easy migration mechanism, to go from one storage format to another.
//
func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
//t.Skip("TODO(jea) bring this back in. broken by the local vs remote shard determination for a cluster")
orig := os.Getenv("PILOSA_TXSRC")
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!
checked := []string{"lmdb", "roaring", "badger", "rbf"}
checked := []string{"lmdb", "roaring", "rbf"}
for _, blue := range checked {
for _, green := range checked {
@ -125,6 +124,7 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
continue
}
blue_green := blue + "_" + green
//vv("setting blue_green to '%v'", blue_green)
// =============================
// Begin setup.
@ -137,6 +137,7 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
t.Fatalf("creating holder: %v", err)
}
defer os.RemoveAll(path)
//vv("path = %v", path)
// we will manually h.Close() below
@ -169,17 +170,26 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
testMustHaveBit(t, h, "i1", "f", 100, 200)
testMustHaveBit(t, h, "i1", "f", 100, 12345678)
//vv("about to reopen; blue_green = '%v' but PILOSA_TXSRC='%v'", blue_green, os.Getenv("PILOSA_TXSRC"))
//h.DumpAllShards()
//vv("after dump, about to close")
h.Close()
//vv("after close, about to re-open")
// can we re.Open the same holder h? hopefully without a problem.
panicOn(h.Open())
testMustHaveBit(t, h, "i0", "f", rowID, colID)
//vv("h.Open() re-open worked; blue_green = '%v'; dump; with PILOSA_TXSRC='%v'", blue_green, os.Getenv("PILOSA_TXSRC"))
//h.DumpAllShards()
testMustHaveBit(t, h, "i0", "f", rowID, colID) // panic here, colID 200 bit was cold.
testMustHaveBit(t, h, "i1", "f", 100, 200)
testMustHaveBit(t, h, "i1", "f", 100, 12345678)
h.Close()
// successful re-open and then Close again of h.
//vv("successful re-open and then Close again of h.")
// check that we can open a NewHolder on green, on same path, and still see our bits.
// Because the NewHolder is the code that creates and configures TxFactory as blue_green.
@ -221,11 +231,14 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
//vv("about to h4.Open we should populate blue from green")
panicOn(h4.Open())
defer h4.Close()
testMustHaveBit(t, h4, "i0", "f", rowID, colID)
testMustHaveBit(t, h4, "i1", "f", 100, 200)
testMustHaveBit(t, h4, "i1", "f", 100, 12345678)
//vv("successfully verified populatingBlueFromGreen with blue_green = '%v'", blue_green)
h4.Close()
os.RemoveAll(path)
}
}
}
@ -234,7 +247,6 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
// go to verify it but blue has more data than green.
// That will also cause query divergence.
func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
//t.Skip("TODO(jea) bring this back in. broken by the local vs remote shard determination for a cluster")
orig := os.Getenv("PILOSA_TXSRC")
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!
@ -260,6 +272,7 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
}
defer os.RemoveAll(path)
//vv("on green, which is '%v'", green)
// we will manually h.Close() below
// Write bits to separate indexes.
@ -297,6 +310,8 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
// open a new holder on path, just looking at blue.
os.Setenv("PILOSA_TXSRC", blue)
//vv("on blue, which is '%v'", blue)
h3 := NewHolder(path, nil)
panicOn(h3.Open())
@ -317,6 +332,8 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
os.Setenv("PILOSA_TXSRC", blue_green)
//vv("on blue_green, which is '%v'", blue_green)
// open a holder with path again, now looking at both blue and green.
// The Holder.Open should do the migration from green, populating blue.
h4 := NewHolder(path, nil)
@ -330,9 +347,15 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
// now open just blue, and add a bit to a new index, i2.
os.Setenv("PILOSA_TXSRC", blue)
//vv("on blue, which is '%v'", blue)
h5 := NewHolder(path, nil)
panicOn(h5.Open())
testSetBit(t, h5, "i2", "f", 500, 777)
//vv("after adding a bit to blue, we have:")
//h5.DumpAllShards()
h5.Close()
// now open blue_green. should get a verification failure
@ -345,6 +368,10 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
// The Holder.Open should verify blue against green and notice the extra bit.
h6 := NewHolder(path, nil)
err = h6.Open()
//vv("h6.Open() had err = '%v', PILOSA_TXSRC='%v'", err, os.Getenv("PILOSA_TXSRC"))
//h6.DumpAllShards()
if err == nil {
h6.Close()
t.Fatalf("should have had blue-green verification fail on Holder.Open")

2
view.go
View file

@ -173,7 +173,7 @@ var workQueue = make(chan struct{}, runtime.NumCPU()*2)
// replaces v.openFragments() with Tx generic code.
func (v *view) openFragmentsInTx() error {
shards, err := v.holder.txf.GetShardsForIndex(v.idx, v.path)
shards, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false)
if err != nil {
return errors.Wrap(err, "DBPerShardGetShardsForIndex()")
}