From 2503fe5b66238751a5031a7da6c923203165b56e Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Fri, 23 Oct 2020 04:54:52 +0000 Subject: [PATCH 1/4] pilosa: avoid re-scanning shards during Holder.Open() - view.openFragmentInTx was forcing a directory scan for shards on every open fragment during Holder.Open(). Seen by pprof profile having excessive allocations from dbshard.go listDirUnderDir(). --- dbshard.go | 175 ++++++++++++++++++++++++++++++--------- dbshard_internal_test.go | 21 ++--- server.go | 4 +- txfactory.go | 28 +++---- version.go | 7 +- view.go | 12 +-- 6 files changed, 169 insertions(+), 78 deletions(-) diff --git a/dbshard.go b/dbshard.go index b4322e1b0..c4fa1a473 100644 --- a/dbshard.go +++ b/dbshard.go @@ -228,6 +228,40 @@ type DBPerShard struct { // roaring doesn't keep a list of open Tx sn. // or default to the 2nd. useOpenList int + + // cache the shards per index to avoid excessive + // directory scans of the index directory. Keep per + // txtype to allow blue-green migrate open to be fast too. + // Keep it up-to-date as we add shards to avoid doing + // a filesystem rescan on new shard creation. + // + // txtype -> index -> *shardSet + index2shards map[txtype]map[string]*shardSet + + isBlueGreen bool +} + +func newIndex2Shards() (r map[txtype]map[string]*shardSet) { + r = make(map[txtype]map[string]*shardSet) + return +} + +type shardSet struct { + shards map[uint64]bool +} + +func (ss *shardSet) Clone() (cp *shardSet) { + cp = newShardSet() + for k, v := range ss.shards { + cp.shards[k] = v + } + return +} + +func newShardSet() *shardSet { + return &shardSet{ + shards: make(map[uint64]bool), + } } // HasData returns true if the database has at least one key. @@ -261,11 +295,11 @@ func (per *DBPerShard) LoadExistingDBs() (err error) { for _, idx := range idxs { - sos, err := per.txf.GetShardsForIndex(idx, "", true) + shardset, err := per.txf.GetShardsForIndex(idx, "", true) if err != nil { return err } - for _, shard := range sos { + for shard := range shardset.shards { _, err := per.GetDBShard(idx.name, shard, idx) if err != nil { return errors.Wrap(err, "DBPerShard.LoadExistingDBs GetDBShard()") @@ -294,14 +328,16 @@ func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string, holder *Ho } d = &DBPerShard{ - types: types, - HolderDir: holderDir, - holder: holder, - dbh: NewDBHolder(), - Flatmap: make(map[flatkey]*DBShard), - txf: txf, - useOpenList: useOpenList, - hasRoaring: hasRoaring, + types: types, + HolderDir: holderDir, + holder: holder, + dbh: NewDBHolder(), + Flatmap: make(map[flatkey]*DBShard), + txf: txf, + useOpenList: useOpenList, + hasRoaring: hasRoaring, + isBlueGreen: len(types) > 1, + index2shards: newIndex2Shards(), } return } @@ -439,6 +475,31 @@ func (per *DBPerShard) prefixForType(idx *Index, ty txtype) string { var ErrNoData = fmt.Errorf("no data") +// keep our cache of shards up-to-date in memory; after the initial +// directory scan, this is all we should we need. Prevents us from +// doing additional, expensive, directory scans. +// +// Caller must hold per.Mu.Lock() already. +func (per *DBPerShard) updateIndex2ShardCacheWithNewShard(dbs *DBShard) { + + for _, ty := range dbs.types { + mapIndex2shardSet, ok := per.index2shards[ty] + if !ok { + mapIndex2shardSet = make(map[string]*shardSet) + per.index2shards[ty] = mapIndex2shardSet + } + // INVAR: mapIndex2shardSet is good, but may be an empty map + + shardset, ok := mapIndex2shardSet[dbs.Index] + if !ok { + shardset = newShardSet() + mapIndex2shardSet[dbs.Index] = shardset + } + // INVAR: shardset is present, not nil; a map that can be added to. + shardset.shards[dbs.Shard] = true + } +} + func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *DBShard, err error) { per.Mu.Lock() @@ -478,6 +539,7 @@ func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs * } dbi.Shard[shard] = dbs + per.updateIndex2ShardCacheWithNewShard(dbs) } if !dbs.Open { var registry DBRegistry @@ -541,22 +603,22 @@ func (per *DBPerShard) Close() (err error) { // DBPerShardGetShardsForIndex returns the shards for idx. // 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) { +func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requireData bool) (*shardSet, error) { - var shards [][]uint64 - for _, ty := range f.types { - var slc []uint64 - slc, err = f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath, requireData) - if err != nil { - return - } - shards = append(shards, slc) - } n := len(f.types) if n != 1 && n != 2 { panic(fmt.Sprintf("internal error. only green or blue/green supported. we see types len %v", n)) } + var shards []*shardSet + for _, ty := range f.types { + ss, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath, requireData) + if err != nil { + return nil, err + } + shards = append(shards, ss) + } + // 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. @@ -569,7 +631,42 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requir // if roaringViewPath is "" then for ty == roaringTxn we go to disk to discover // all the view paths under idx for type ty. // 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) { +// The returned sliceOfShards should not be modified. We will cache it for subsequent +// queries. +// +// when a new DBShard is made, we will update the list of shards then. Thus +// the per.index2shard should always be up to date AFTER the first call here. +func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (setOfShards *shardSet, err error) { + + // use the cache, always + per.Mu.Lock() + defer per.Mu.Unlock() + + i2ss, ok := per.index2shards[ty] + if !ok { + // index -> shardSet + i2ss = make(map[string]*shardSet) + per.index2shards[ty] = i2ss + } + // INVAR: i2ss is good, but may be an empty map + + ss, ok := i2ss[idx.name] + if ok { + return ss.Clone(), nil + } + // INVAR: cache miss, and index2shards[ty] exists. + + // Upon return, cache the setOfShards value and reuse it next time. + defer func() { + per.index2shards[ty][idx.name] = setOfShards + + // don't return the actual cache, or we'll get read/write to map races. + setOfShards = setOfShards.Clone() + }() + + // gotta read shards from disk directory layout. + + setOfShards = newShardSet() if ty == roaringTxn { rx := &RoaringTx{ @@ -585,13 +682,24 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r errors.Wrap(err, fmt.Sprintf( "TypedDBPerShardGetLocalShardsForIndex roaringTxn view.path='%v'", view.path)) } - sliceOfShards = append(sliceOfShards, sos...) + for _, shard := range sos { + setOfShards.shards[shard] = true + } } } - return dedupShardSlice(sliceOfShards), nil + return setOfShards, nil } - return rx.SliceOfShards("", "", "", roaringViewPath) + sos, err := rx.SliceOfShards("", "", "", roaringViewPath) + if err != nil { + return nil, err + } + for _, shard := range sos { + setOfShards.shards[shard] = true + } + return setOfShards, nil } + // INVAR: not-roaring. + requiredSuffix := ty.FileSuffix() path := per.prefixForType(idx, ty) @@ -631,14 +739,14 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r return nil, err } if hasData { - sliceOfShards = append(sliceOfShards, shard) + setOfShards.shards[shard] = true } } else { // file presence is enough - sliceOfShards = append(sliceOfShards, shard) + setOfShards.shards[shard] = true } } - return + return setOfShards, nil } func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) { @@ -663,10 +771,10 @@ func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint6 } func listDirUnderDir(root string, includeRoot bool, requiredSuffix string, ignoreEmpty bool) (files []string, err error) { - if !dirExists(root) { return } + n := len(root) + 1 if includeRoot { n = 0 @@ -881,14 +989,3 @@ func (dbs *DBShard) verifyBlueEqualsGreen() (err error) { return nil } - -func dedupShardSlice(sos []uint64) (r []uint64) { - m := make(map[uint64]struct{}) - for _, s := range sos { - m[s] = struct{}{} - } - for k := range m { - r = append(r, k) - } - return -} diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index 7cbcb95da..bb2beac3e 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -88,21 +88,21 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { estd := "rick/_exists/views/standard" std := "rick/f/views/standard" - sos, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false) + shardset, 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)) + if !shardset.shards[shard] { + panic(fmt.Sprintf("missing shard=%v from shardset='%#v'", shard, shardset.shards)) } } if src == "roaring" { // check estd too - sos, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false) + shardset, 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) { - panic(fmt.Sprintf("missing shard=%v from sos='%#v'", shard, sos)) + if !shardset.shards[shard] { + panic(fmt.Sprintf("missing shard=%v from shardset='%#v'", shard, shardset.shards)) } } } @@ -110,15 +110,6 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { } } -func inSlice(sos []uint64, shard uint64) bool { - for i := range sos { - if shard == sos[i] { - return true - } - } - return false -} - // data for Test_DBPerShard_GetShardsForIndex // var sampleRoaringDirList = map[string]string{"roaring": ` diff --git a/server.go b/server.go index 70123465a..b1fa62e2b 100644 --- a/server.go +++ b/server.go @@ -462,7 +462,7 @@ func (s *Server) InternalClient() InternalClient { // UpAndDown brings the server up minimally and shuts it down // again; basically, it exists for testing holder open and close. func (s *Server) UpAndDown() error { - s.logger.Printf("open server") + s.logger.Printf("open server. PID %v", os.Getpid()) // Log startup err := s.holder.logStartup() @@ -485,7 +485,7 @@ func (s *Server) UpAndDown() error { // Open opens and initializes the server. func (s *Server) Open() error { - s.logger.Printf("open server") + s.logger.Printf("open server. PID %v", os.Getpid()) if s.holder.NeedsSnapshot() { // Start background monitoring. diff --git a/txfactory.go b/txfactory.go index da0b9497b..9a470f1a7 100644 --- a/txfactory.go +++ b/txfactory.go @@ -1305,20 +1305,24 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) { } if verifyInsteadOfCopy { - diff := f.shardSliceDiff(blueShards, greenShards) + diff := f.shardSetDiff(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) + meta := newShardSet() + for _, shard := range shards { + meta.shards[shard] = true + } + diff2 := f.shardSetDiff(greenShards, meta) 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) } } - for _, shard := range greenShards { + for shard := range greenShards.shards { dbs, err := f.dbPerShard.GetDBShard(idx.name, shard, idx) if err != nil { @@ -1349,22 +1353,14 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) { return nil } -func (f *TxFactory) shardSliceDiff(blueShards, greenShards []uint64) (diff string) { - nb := len(blueShards) - ng := len(greenShards) +func (f *TxFactory) shardSetDiff(blueShards, greenShards *shardSet) (diff string) { + nb := len(blueShards.shards) + ng := len(greenShards.shards) if nb != ng { diff = fmt.Sprintf("blueShard[%v] count = %v; greenShard[%v] count = %v; ", f.types[0], nb, f.types[1], ng) } - b := make(map[uint64]bool) - g := make(map[uint64]bool) - for _, bs := range blueShards { - b[bs] = true - } - for _, gs := range greenShards { - g[gs] = true - } - bmg := mapDiff(b, g) // get blue - green - gmb := mapDiff(g, b) // get green - blue + bmg := mapDiff(blueShards.shards, greenShards.shards) // get blue - green + gmb := mapDiff(greenShards.shards, blueShards.shards) // get green - blue if len(bmg) == 0 && len(gmb) == 0 { return "" diff --git a/version.go b/version.go index 4313c9e33..91a5841b7 100644 --- a/version.go +++ b/version.go @@ -14,13 +14,17 @@ package pilosa -import "time" +import ( + "runtime" + "time" +) var Version string var Commit string var Variant string var BuildTime string var LatticeCommit string +var GoVersion string = runtime.Version() func VersionInfo() string { var prefix string @@ -48,6 +52,7 @@ func VersionInfo() string { case buildTime != "": suffix += " (" + buildTime + ")" } + suffix += " " + GoVersion return prefix + "Pilosa" + suffix } diff --git a/view.go b/view.go index 65f73d3c4..d59f041be 100644 --- a/view.go +++ b/view.go @@ -173,20 +173,22 @@ 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, false) + shardSet, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false) if err != nil { return errors.Wrap(err, "DBPerShardGetShardsForIndex()") } + shards := shardSet.shards eg, ctx := errgroup.WithContext(context.Background()) var mu sync.Mutex - shardCh := make(chan uint64, len(shards)) - for i := range shards { - shardCh <- shards[i] + n := len(shards) + shardCh := make(chan uint64, n) + for shard := range shards { + shardCh <- shard } shardLoop: - for range shards { + for j := 0; j < n; j++ { select { case <-ctx.Done(): break shardLoop From 8ff6e8e0fd428a40a3198547ee6a029f451b0bdb Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Fri, 23 Oct 2020 23:49:34 +0000 Subject: [PATCH 2/4] versioned readonly shards map --- dbshard.go | 63 +++++++++++++++++++++++++--------------- dbshard_internal_test.go | 12 ++++---- txfactory.go | 16 +++++----- view.go | 8 ++--- 4 files changed, 57 insertions(+), 42 deletions(-) diff --git a/dbshard.go b/dbshard.go index c4fa1a473..ff8ae5c53 100644 --- a/dbshard.go +++ b/dbshard.go @@ -247,15 +247,36 @@ func newIndex2Shards() (r map[txtype]map[string]*shardSet) { } type shardSet struct { - shards map[uint64]bool + shards map[uint64]bool + shardsVer int64 // increment with each change. + + // give out readonly to repeated consumers if + // readonlyVer == shardsVer + readonly map[uint64]bool + readonlyVer int64 } -func (ss *shardSet) Clone() (cp *shardSet) { - cp = newShardSet() - for k, v := range ss.shards { - cp.shards[k] = v +// CloneMaybe maintains a re-usable readonly version +// ss.shards that can be returned to multiple goroutine +// reads as it will never change. A copy is only made +// once for each change in the shard set. +func (ss *shardSet) CloneMaybe() map[uint64]bool { + + if ss.readonlyVer == ss.shardsVer { + return ss.readonly } - return + + // readonlyVer is out of date. + // readonly needs update. We cannot + // modify the readonly map in place; + // must make a fully new copy here. + ss.readonly = make(map[uint64]bool) + + for k, v := range ss.shards { + ss.readonly[k] = v + } + ss.readonlyVer = ss.shardsVer + return ss.readonly } func newShardSet() *shardSet { @@ -299,7 +320,7 @@ func (per *DBPerShard) LoadExistingDBs() (err error) { if err != nil { return err } - for shard := range shardset.shards { + for shard := range shardset { _, err := per.GetDBShard(idx.name, shard, idx) if err != nil { return errors.Wrap(err, "DBPerShard.LoadExistingDBs GetDBShard()") @@ -497,6 +518,7 @@ func (per *DBPerShard) updateIndex2ShardCacheWithNewShard(dbs *DBShard) { } // INVAR: shardset is present, not nil; a map that can be added to. shardset.shards[dbs.Shard] = true + shardset.shardsVer++ // invalid the readonly copy, force cloning it anew. } } @@ -603,14 +625,14 @@ func (per *DBPerShard) Close() (err error) { // DBPerShardGetShardsForIndex returns the shards for idx. // 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) (*shardSet, error) { +func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requireData bool) (map[uint64]bool, error) { n := len(f.types) if n != 1 && n != 2 { panic(fmt.Sprintf("internal error. only green or blue/green supported. we see types len %v", n)) } - var shards []*shardSet + var shards []map[uint64]bool for _, ty := range f.types { ss, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath, requireData) if err != nil { @@ -636,7 +658,7 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requir // // when a new DBShard is made, we will update the list of shards then. Thus // the per.index2shard should always be up to date AFTER the first call here. -func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (setOfShards *shardSet, err error) { +func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (shardMap map[uint64]bool, err error) { // use the cache, always per.Mu.Lock() @@ -652,21 +674,16 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r ss, ok := i2ss[idx.name] if ok { - return ss.Clone(), nil + return ss.CloneMaybe(), nil } // INVAR: cache miss, and index2shards[ty] exists. - // Upon return, cache the setOfShards value and reuse it next time. - defer func() { - per.index2shards[ty][idx.name] = setOfShards - - // don't return the actual cache, or we'll get read/write to map races. - setOfShards = setOfShards.Clone() - }() - // gotta read shards from disk directory layout. + setOfShards := newShardSet() + per.index2shards[ty][idx.name] = setOfShards + setOfShards.shardsVer++ // invalidate readonlyVer - setOfShards = newShardSet() + // Upon return, cache the setOfShards value and reuse it next time if ty == roaringTxn { rx := &RoaringTx{ @@ -687,7 +704,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r } } } - return setOfShards, nil + return setOfShards.CloneMaybe(), nil } sos, err := rx.SliceOfShards("", "", "", roaringViewPath) if err != nil { @@ -696,7 +713,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r for _, shard := range sos { setOfShards.shards[shard] = true } - return setOfShards, nil + return setOfShards.CloneMaybe(), nil } // INVAR: not-roaring. @@ -746,7 +763,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r setOfShards.shards[shard] = true } } - return setOfShards, nil + return setOfShards.CloneMaybe(), nil } func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) { diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index bb2beac3e..52b9b3027 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -88,21 +88,21 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { estd := "rick/_exists/views/standard" std := "rick/f/views/standard" - shardset, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false) + shards, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false) panicOn(err) for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { - if !shardset.shards[shard] { - panic(fmt.Sprintf("missing shard=%v from shardset='%#v'", shard, shardset.shards)) + if !shards[shard] { + panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards)) } } if src == "roaring" { // check estd too - shardset, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false) + shards, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false) panicOn(err) for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { - if !shardset.shards[shard] { - panic(fmt.Sprintf("missing shard=%v from shardset='%#v'", shard, shardset.shards)) + if !shards[shard] { + panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards)) } } } diff --git a/txfactory.go b/txfactory.go index 9a470f1a7..6e9efe1fb 100644 --- a/txfactory.go +++ b/txfactory.go @@ -1312,9 +1312,9 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) { // can also check against meta data shards := idx.AvailableShards(localOnly).Slice() - meta := newShardSet() + meta := make(map[uint64]bool) for _, shard := range shards { - meta.shards[shard] = true + meta[shard] = true } diff2 := f.shardSetDiff(greenShards, meta) if diff2 != "" { @@ -1322,7 +1322,7 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) { } } - for shard := range greenShards.shards { + for shard := range greenShards { dbs, err := f.dbPerShard.GetDBShard(idx.name, shard, idx) if err != nil { @@ -1353,14 +1353,14 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) { return nil } -func (f *TxFactory) shardSetDiff(blueShards, greenShards *shardSet) (diff string) { - nb := len(blueShards.shards) - ng := len(greenShards.shards) +func (f *TxFactory) shardSetDiff(blueShards, greenShards map[uint64]bool) (diff string) { + nb := len(blueShards) + ng := len(greenShards) if nb != ng { diff = fmt.Sprintf("blueShard[%v] count = %v; greenShard[%v] count = %v; ", f.types[0], nb, f.types[1], ng) } - bmg := mapDiff(blueShards.shards, greenShards.shards) // get blue - green - gmb := mapDiff(greenShards.shards, blueShards.shards) // get green - blue + bmg := mapDiff(blueShards, greenShards) // get blue - green + gmb := mapDiff(greenShards, blueShards) // get green - blue if len(bmg) == 0 && len(gmb) == 0 { return "" diff --git a/view.go b/view.go index d59f041be..ef39de39a 100644 --- a/view.go +++ b/view.go @@ -173,22 +173,20 @@ var workQueue = make(chan struct{}, runtime.NumCPU()*2) // replaces v.openFragments() with Tx generic code. func (v *view) openFragmentsInTx() error { - shardSet, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false) + shards, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false) if err != nil { return errors.Wrap(err, "DBPerShardGetShardsForIndex()") } - shards := shardSet.shards eg, ctx := errgroup.WithContext(context.Background()) var mu sync.Mutex - n := len(shards) - shardCh := make(chan uint64, n) + shardCh := make(chan uint64, len(shards)) for shard := range shards { shardCh <- shard } shardLoop: - for j := 0; j < n; j++ { + for range shards { select { case <-ctx.Done(): break shardLoop From c5e46e4618c007fcfc962860b89ab5d84c6fd0ba Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Sat, 24 Oct 2020 00:20:30 +0000 Subject: [PATCH 3/4] one copy of shard map during a reload --- dbshard.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/dbshard.go b/dbshard.go index ff8ae5c53..ede2959f0 100644 --- a/dbshard.go +++ b/dbshard.go @@ -266,6 +266,8 @@ func (ss *shardSet) CloneMaybe() map[uint64]bool { return ss.readonly } + //vv("888888 CloneMaybe updating to version %v with %v shards", ss.shardsVer, len(ss.shards)) + // readonlyVer is out of date. // readonly needs update. We cannot // modify the readonly map in place; @@ -515,10 +517,15 @@ func (per *DBPerShard) updateIndex2ShardCacheWithNewShard(dbs *DBShard) { if !ok { shardset = newShardSet() mapIndex2shardSet[dbs.Index] = shardset + //vv("888888 made new shardset for shard dbs.Shard=%v", dbs.Shard) } // INVAR: shardset is present, not nil; a map that can be added to. - shardset.shards[dbs.Shard] = true - shardset.shardsVer++ // invalid the readonly copy, force cloning it anew. + _, already := shardset.shards[dbs.Shard] + if !already { + shardset.shards[dbs.Shard] = true + shardset.shardsVer++ // invalid the readonly copy, force cloning it anew. + //vv("888888 updated to shardset version %v with %v shards", shardset.shardsVer, len(shardset.shards)) + } } } From 233b3cbc0fbcb9d6c9ee716b4745f4d97907c09c Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Sat, 24 Oct 2020 00:35:25 +0000 Subject: [PATCH 4/4] versioned map cleanup --- dbshard.go | 27 +++++++++++++-------------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/dbshard.go b/dbshard.go index ede2959f0..a0912c732 100644 --- a/dbshard.go +++ b/dbshard.go @@ -256,6 +256,14 @@ type shardSet struct { readonlyVer int64 } +func (ss *shardSet) add(shard uint64) { + _, already := ss.shards[shard] + if !already { + ss.shards[shard] = true + ss.shardsVer++ + } +} + // CloneMaybe maintains a re-usable readonly version // ss.shards that can be returned to multiple goroutine // reads as it will never change. A copy is only made @@ -266,8 +274,6 @@ func (ss *shardSet) CloneMaybe() map[uint64]bool { return ss.readonly } - //vv("888888 CloneMaybe updating to version %v with %v shards", ss.shardsVer, len(ss.shards)) - // readonlyVer is out of date. // readonly needs update. We cannot // modify the readonly map in place; @@ -517,15 +523,9 @@ func (per *DBPerShard) updateIndex2ShardCacheWithNewShard(dbs *DBShard) { if !ok { shardset = newShardSet() mapIndex2shardSet[dbs.Index] = shardset - //vv("888888 made new shardset for shard dbs.Shard=%v", dbs.Shard) } // INVAR: shardset is present, not nil; a map that can be added to. - _, already := shardset.shards[dbs.Shard] - if !already { - shardset.shards[dbs.Shard] = true - shardset.shardsVer++ // invalid the readonly copy, force cloning it anew. - //vv("888888 updated to shardset version %v with %v shards", shardset.shardsVer, len(shardset.shards)) - } + shardset.add(dbs.Shard) } } @@ -688,7 +688,6 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r // gotta read shards from disk directory layout. setOfShards := newShardSet() per.index2shards[ty][idx.name] = setOfShards - setOfShards.shardsVer++ // invalidate readonlyVer // Upon return, cache the setOfShards value and reuse it next time @@ -707,7 +706,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r "TypedDBPerShardGetLocalShardsForIndex roaringTxn view.path='%v'", view.path)) } for _, shard := range sos { - setOfShards.shards[shard] = true + setOfShards.add(shard) } } } @@ -718,7 +717,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r return nil, err } for _, shard := range sos { - setOfShards.shards[shard] = true + setOfShards.add(shard) } return setOfShards.CloneMaybe(), nil } @@ -763,11 +762,11 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r return nil, err } if hasData { - setOfShards.shards[shard] = true + setOfShards.add(shard) } } else { // file presence is enough - setOfShards.shards[shard] = true + setOfShards.add(shard) } } return setOfShards.CloneMaybe(), nil